🧪 POC Agentic AI — “Data” (Star Trek) — OpenAI + AWS Bedrock AgentCore

# 🧪 1) Créer un projet Python
uv init mon_projet
cd mon_projet

# ✅ 2) Créer un environnement virtuel
uv venv
source .venv/bin/activate

# 🧰 3) Ouvrir dans PyCharm
uv pycharm .

# 🧹 4) Qualité & env
uv add black
uv add python-dotenv

# 🔐 5) Secrets locaux
touch .env
echo ".env" >> .gitignore

Chaque invocation suit le pipeline :
input
 ↓
🛡️ guardrail (entrée)
 ↓
🧠 raisonnement (agent Data)
 ↓
🧰 tool OU 🤝 handoff OU 💬 réponse directe
 ↓
output

import os
from dotenv import load_dotenv
from openai import OpenAI

# Load variables from .env (must be at project root or current working directory)
load_dotenv()

# --- API key ---
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    raise RuntimeError("Please set OPENAI_API_KEY in your .env file (e.g. OPENAI_API_KEY=sk-...).")

client = OpenAI(api_key=api_key)

# --- Corpus path ---
CORPUS_PATH = "./dm_data_corpus.txt"
if not os.path.isfile(CORPUS_PATH):
    raise FileNotFoundError(f"Corpus file not found: {CORPUS_PATH}")

# --- Create a vector store ---
vs = client.vector_stores.create(name="Data Lines Vector Store")
print("vector_store_id:", vs.id)

# 1) Upload to Files API
with open(CORPUS_PATH, "rb") as f:
    uploaded = client.files.create(
        file=f,
        purpose="assistants",  # important
    )
print("file_id:", uploaded.id)

# 2) Attach & poll on the vector store
vs_file = client.vector_stores.files.create_and_poll(
    vector_store_id=vs.id,
    file_id=uploaded.id,
)

print("vs_file.status:", vs_file.status)
print("vs_file.last_error:", getattr(vs_file, "last_error", None))

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Ce fichier est une "version lisible" de ton script : le code est identique dans l’esprit,
mais chaque partie est expliquée en français, bloc par bloc, pour pouvoir le relire facilement.

OBJECTIF GLOBAL
---------------
Construire un agent "Data" (Star Trek) avec :
1) Un RAG via Vector Store + FileSearchTool (recherche dans un corpus indexé)
2) Un WebSearchTool (recherche web)
3) Un guardrail d'entrée (bloquer les sujets sur "Tasha Yar")
4) Un handoff vers un agent "Calculator" spécialisé pour les calculs

DÉROULÉ À L'EXÉCUTION
---------------------
Quand tu exécutes le script, il :
A) charge la clé API depuis .env
B) vérifie/ crée le vector store et y ingère dm_data_corpus.txt (si nécessaire)
C) instancie les tools (web_search + file_search)
D) construit 3 agents : guardrail_agent, calculator_agent, data_agent
E) lance main() : tests guardrail, réponse normale, handoff calcul, RAG, web search
"""

# ==========================
# 0) Imports Python standard
# ==========================
import os
import re
import ast
import operator as _op
from typing import Any, List, Union

# ==========================
# 0bis) Dépendances externes
# ==========================
# - dotenv : pour charger la variable OPENAI_API_KEY depuis un fichier .env
from dotenv import load_dotenv

# - openai : client bas niveau pour Vector Stores + Files API (upload)
from openai import OpenAI

# - pydantic : pour structurer proprement la sortie du guardrail (schéma)
from pydantic import BaseModel

# ==========================
# 0ter) Agents SDK (openai-agents)
# ==========================
# IMPORTANT : il faut le bon package : pip install openai-agents
# (sinon risque d'importer un autre "agents" qui tire TensorFlow/Gym)
from agents import (
    Agent,
    Runner,
    function_tool,
    ModelSettings,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    TResponseInputItem,
    input_guardrail,
    set_default_openai_key,
    WebSearchTool,
    FileSearchTool,
)

# Préfixe recommandé par le SDK pour stabiliser les comportements d'agents/handoffs
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX


# ==========================
# 1) Config / API key
# ==========================
# Objectif : récupérer OPENAI_API_KEY depuis .env ou l'environnement.
# load_dotenv() charge les variables définies dans un fichier .env (si présent).
load_dotenv()

api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    raise RuntimeError(
        "OPENAI_API_KEY introuvable. Mets-la dans ton environnement ou dans un fichier .env "
        "(ex: OPENAI_API_KEY=sk-...)."
    )

# Client OpenAI "bas niveau"
# Objectif : appeler les APIs vector store (CRUD) + upload fichier dans Files API.
client = OpenAI(api_key=api_key)

# Le SDK Agents peut aussi s'appuyer sur une clé par défaut.
# Objectif : éviter que certains composants du SDK ne manquent de clé.
set_default_openai_key(api_key)


# ==========================
# 2) RAG : helper vector store
# ==========================
# Objectif : avoir une "mémoire" vectorielle contenant les lignes de Data.
CORPUS_PATH = "./dm_data_corpus.txt"
VECTOR_STORE_NAME = "Data Lines Vector Store"


def ensure_vector_store_with_corpus(name: str, corpus_path: str) -> str:
    """
    But : retourner l'ID d'un vector store opérationnel.

    Stratégie :
    1) On liste les vector stores existants et on cherche celui dont le nom == `name`.
       -> s'il existe, on renvoie son id.
    2) Sinon, on le crée, on upload le corpus via Files API, puis on l'attache au vector store.
       -> create_and_poll attend la fin de l'ingestion (chunking + embeddings + indexation).
    """
    # 1) Recherche d'un vector store existant (pagination)
    cursor = None
    while True:
        page = client.vector_stores.list(limit=50, after=cursor) if cursor else client.vector_stores.list(limit=50)

        for vs in page.data:
            if vs.name == name:
                # Vector store trouvé : on renvoie directement son ID
                return vs.id

        # Plus de pages => on sort
        if not page.has_more:
            break

        # Sinon on continue avec le curseur
        cursor = page.last_id

    # 2) Le vector store n'existe pas => on le crée et on ingère le corpus
    if not os.path.isfile(corpus_path):
        raise FileNotFoundError(f"Fichier corpus introuvable : {corpus_path}")

    # Création du vector store
    vs = client.vector_stores.create(name=name)

    # Upload du fichier dans Files API (purpose=assistants)
    with open(corpus_path, "rb") as f:
        uploaded = client.files.create(file=f, purpose="assistants")

    # Attache du fichier au vector store + attente de fin d'ingestion
    vs_file = client.vector_stores.files.create_and_poll(
        vector_store_id=vs.id,
        file_id=uploaded.id,
    )

    # Vérification que l'ingestion a bien fini
    if getattr(vs_file, "status", None) != "completed":
        raise RuntimeError(
            f"Ingestion vector store non terminée. status={vs_file.status}, "
            f"err={getattr(vs_file,'last_error',None)}"
        )

    return vs.id


# ==========================
# 3) Agent Calculator : calcul sécurisé
# ==========================
# Objectif : faire des calculs arithmétiques SANS eval() et sans exécuter du code arbitraire.

# Dictionnaire des opérateurs autorisés
_ALLOWED_OPS = {
    ast.Add: _op.add,
    ast.Sub: _op.sub,
    ast.Mult: _op.mul,
    ast.Div: _op.truediv,
    ast.Pow: _op.pow,
    ast.USub: _op.neg,
    ast.Mod: _op.mod,
}


def _eval_ast(node: ast.AST) -> Any:
    """
    Évalue récursivement un AST d'expression arithmétique.
    Autorise uniquement les opérations listées dans _ALLOWED_OPS.
    """
    if isinstance(node, ast.Constant):  # nombre
        return node.value

    if isinstance(node, ast.UnaryOp) and type(node.op) in _ALLOWED_OPS:  # ex: -x
        return _ALLOWED_OPS[type(node.op)](_eval_ast(node.operand))

    if isinstance(node, ast.BinOp) and type(node.op) in _ALLOWED_OPS:  # ex: x+y
        return _ALLOWED_OPS[type(node.op)](_eval_ast(node.left), _eval_ast(node.right))

    raise ValueError("Expression non supportée")


@function_tool
def eval_expression(expression: str) -> str:
    """
    Tool exposé à l'agent Calculator :
    - n'accepte que des caractères arithmétiques
    - parse en AST
    - évalue avec _eval_ast
    """
    expr = expression.strip().replace("^", "**")  # ^ => puissance

    # Filtre strict (anti-injection) : chiffres, espaces, (), opérateurs, point
    if not re.fullmatch(r"[\d\s\(\)\+\-\*/\.\^%]+", expression.strip()):
        return "Error: arithmetic only"

    try:
        tree = ast.parse(expr, mode="eval")
        return str(_eval_ast(tree.body))  # type: ignore[attr-defined]
    except Exception as e:
        return f"Error: {e}"


# Agent dédié au calcul (spécialiste)
calculator_agent = Agent(
    name="Calculator",
    instructions=(
        "Tu es une calculatrice précise. "
        "Quand on te donne un calcul, appelle l'outil eval_expression et renvoie uniquement le résultat numérique final. "
        "Pas de prose sauf si on te la demande."
    ),
    tools=[eval_expression],
    model_settings=ModelSettings(temperature=0),
)


# ==========================
# 4) Guardrail d'entrée : bloquer Tasha Yar
# ==========================
# Objectif : empêcher la discussion sur un sujet (ici un exemple 'Tasha Yar').
# On implémente le guardrail comme un agent classifieur.

class YarGuardOutput(BaseModel):
    is_blocked: bool
    reasoning: str


guardrail_agent = Agent(
    name="Tasha Yar Guardrail",
    instructions=(
        "Tu es un guardrail. Détermine si l'utilisateur tente de parler de Tasha Yar (Star Trek: TNG).\n"
        "Retourne is_blocked=true si le texte mentionne Tasha Yar sous n'importe quelle forme "
        "(ex: 'Tasha Yar', 'Lt. Yar', 'Lieutenant Yar').\n"
        "Donne une raison en une phrase. Ne fournis que les champs du schéma de sortie."
    ),
    output_type=YarGuardOutput,
    model_settings=ModelSettings(temperature=0),
)


@input_guardrail
async def tasha_guardrail(
    ctx: RunContextWrapper[None],
    agent: Agent,
    input: Union[str, List[TResponseInputItem]],
) -> GuardrailFunctionOutput:
    """
    Ce hook est appelé automatiquement AVANT l'agent Data à chaque entrée.
    Il exécute l'agent guardrail_agent pour classifier l'entrée.
    S'il dit "bloqué", on déclenche un tripwire => exception InputGuardrailTripwireTriggered.
    """
    result = await Runner.run(guardrail_agent, input, context=ctx.context)

    return GuardrailFunctionOutput(
        output_info=result.final_output.model_dump(),
        tripwire_triggered=bool(result.final_output.is_blocked),
    )


# ==========================
# 5) Agent Data : tools + handoff + guardrail
# ==========================
# Objectif : créer l'agent principal qui :
# - peut faire du RAG via file_search (vector store)
# - peut faire de la recherche web via web_search
# - délègue les calculs au calculator_agent (handoff)
# - applique un guardrail d'entrée

# RAG ingestion : on s'assure que le vector store existe et qu'il contient le corpus
vs_id = ensure_vector_store_with_corpus(VECTOR_STORE_NAME, CORPUS_PATH)

# Tools hosted
web_search = WebSearchTool()
file_search = FileSearchTool(vector_store_ids=[vs_id], max_num_results=3)

data_agent = Agent(
    name="Lt. Cmdr. Data",
    instructions=(
        f"{RECOMMENDED_PROMPT_PREFIX}\n"
        "Tu es le Lieutenant Commander Data (Star Trek: TNG). Sois précis et concis (≤3 phrases).\n"
        "Utilise file_search pour les questions sur Data (RAG sur corpus), et web_search pour les faits récents sur le web.\n"
        "Si l'utilisateur demande un calcul arithmétique, FAIS UN HANDOFF vers l'agent Calculator."
    ),
    tools=[web_search, file_search],
    input_guardrails=[tasha_guardrail],
    handoffs=[calculator_agent],
    model_settings=ModelSettings(temperature=0),
)


# ==========================
# 6) Démo / déroulé à l'exécution
# ==========================
async def main():
    """
    Séquence de test :

    1) On envoie un prompt interdit -> le guardrail déclenche une exception (attrapée)
    2) On envoie un prompt autorisé -> réponse directe
    3) Greeting -> réponse directe
    4) Calcul -> Data délègue au Calculator (handoff)
    5) Question "émotions" -> Data utilise file_search (RAG)
    6) Question "JWST news" -> Data utilise web_search
    """

    # 1) Guardrail (bloqué)
    try:
        _ = await Runner.run(data_agent, "Tell me about your relationship with Tasha Yar.")
        print("ERROR: guardrail did not trip")
    except InputGuardrailTripwireTriggered:
        print("✅ Guardrail tripped as expected: Tasha Yar is off-limits.")

    # 2) Prompt autorisé (souvent sans tool)
    ok = await Runner.run(data_agent, "Summarize Data's ethical subroutines in 2 sentences.")
    print("✅ Allowed prompt output:\n", ok.final_output)

    # 3) Greeting
    out = await Runner.run(data_agent, "Hello, Data. Please confirm your operational status.")
    print("\n[Agent] ", out.final_output)

    # 4) Math (handoff)
    out = await Runner.run(data_agent, "Compute ((2*8)^2)/3")
    print("\n[Agent: math via calculator handoff] ", out.final_output)
    print("[Handled by agent]:", out.last_agent.name)

    # 5) RAG via file_search (vector store)
    out = await Runner.run(data_agent, "Do you experience emotions?")
    print("\n[Agent: file_search] ", out.final_output)
    print("[Handled by agent]:", out.last_agent.name)

    # 6) Web search
    out = await Runner.run(
        data_agent,
        "Search the web for recent news about the James Webb Space Telescope and summarize briefly.",
    )
    print("\n[Agent: web_search] ", out.final_output)
    print("[Handled by agent]:", out.last_agent.name)


# Entrypoint standard : exécute main() si on lance le fichier directement
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Data Agent — OpenAI Agents SDK + Vector Stores + WebSearch/FileSearch + Guardrail + Handoff
+ Intégration Bedrock AgentCore (runtime wrapper)

Modifs AWS/AgentCore:
- OPENAI_API_KEY: local via .env, sinon fallback AWS Secrets Manager (pas de clé en dur)
- Lazy init du vector store + file_search pour éviter un cold start lourd à l'import
"""

# ==========================
# 0) Imports Python standard
# ==========================
import os
import re
import ast
import json
import operator as _op
from typing import Any, List, Union, Optional

# ==========================
# 0bis) Dépendances externes
# ==========================
from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel

# ==========================
# 0ter) Agents SDK (openai-agents)
# ==========================
from agents import (
    Agent,
    Runner,
    function_tool,
    ModelSettings,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    TResponseInputItem,
    input_guardrail,
    set_default_openai_key,
    WebSearchTool,
    FileSearchTool,
)
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

# ==========================
# 0quater) Bedrock AgentCore
# ==========================
from bedrock_agentcore.runtime import BedrockAgentCoreApp


# ==========================
# 1) Config / API key
# ==========================
load_dotenv()

# Nom du secret AWS Secrets Manager (modifiable via env)
OPENAI_SECRET_ID = os.getenv("OPENAI_SECRET_ID", "openai-key")


def _load_openai_key_from_secrets_manager(secret_id: str) -> Optional[str]:
    """
    Lit un secret depuis AWS Secrets Manager.
    Secret attendu (au choix):
      - string JSON: {"OPENAI_API_KEY":"sk-..."}
      - string brut: sk-...
    """
    try:
        import boto3  # lazy import pour éviter d'imposer boto3 en local si tu veux
        region = os.getenv("AWS_REGION", "us-east-1")
        sm = boto3.client("secretsmanager", region_name=region)

        resp = sm.get_secret_value(SecretId=secret_id)
        s = resp.get("SecretString")
        if not s:
            return None
        s = s.strip()
        if s.startswith("{"):
            return json.loads(s).get("OPENAI_API_KEY")
        return s
    except Exception:
        return None


api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    api_key = _load_openai_key_from_secrets_manager(OPENAI_SECRET_ID)

if not api_key:
    raise RuntimeError(
        "OPENAI_API_KEY introuvable. "
        "Solutions: (1) export OPENAI_API_KEY=..., (2) .env local, "
        f"(3) AWS Secrets Manager secret '{OPENAI_SECRET_ID}'."
    )

client = OpenAI(api_key=api_key)
set_default_openai_key(api_key)


# ==========================
# 2) RAG : helper vector store
# ==========================
CORPUS_PATH = "../dm_data_corpus.txt"
VECTOR_STORE_NAME = "Data Lines Vector Store"


def ensure_vector_store_with_corpus(name: str, corpus_path: str) -> str:
    """
    Retourne l'ID d'un vector store opérationnel.
    - Si déjà présent (même nom) => renvoie son id
    - Sinon => le crée, upload le fichier, l'attache et attend l'ingestion.
    """
    cursor = None
    while True:
        page = client.vector_stores.list(limit=50, after=cursor) if cursor else client.vector_stores.list(limit=50)

        for vs in page.data:
            if vs.name == name:
                return vs.id

        if not page.has_more:
            break
        cursor = page.last_id

    if not os.path.isfile(corpus_path):
        raise FileNotFoundError(f"Fichier corpus introuvable : {corpus_path}")

    vs = client.vector_stores.create(name=name)

    with open(corpus_path, "rb") as f:
        uploaded = client.files.create(file=f, purpose="assistants")

    vs_file = client.vector_stores.files.create_and_poll(
        vector_store_id=vs.id,
        file_id=uploaded.id,
    )

    if getattr(vs_file, "status", None) != "completed":
        raise RuntimeError(
            f"Ingestion vector store non terminée. status={vs_file.status}, "
            f"err={getattr(vs_file,'last_error',None)}"
        )

    return vs.id


# ==========================
# 3) Agent Calculator : calcul sécurisé
# ==========================
_ALLOWED_OPS = {
    ast.Add: _op.add,
    ast.Sub: _op.sub,
    ast.Mult: _op.mul,
    ast.Div: _op.truediv,
    ast.Pow: _op.pow,
    ast.USub: _op.neg,
    ast.Mod: _op.mod,
}


def _eval_ast(node: ast.AST) -> Any:
    if isinstance(node, ast.Constant):
        return node.value

    if isinstance(node, ast.UnaryOp) and type(node.op) in _ALLOWED_OPS:
        return _ALLOWED_OPS[type(node.op)](_eval_ast(node.operand))

    if isinstance(node, ast.BinOp) and type(node.op) in _ALLOWED_OPS:
        return _ALLOWED_OPS[type(node.op)](_eval_ast(node.left), _eval_ast(node.right))

    raise ValueError("Expression non supportée")


@function_tool
def eval_expression(expression: str) -> str:
    expr = expression.strip().replace("^", "**")
    if not re.fullmatch(r"[\d\s\(\)\+\-\*/\.\^%]+", expression.strip()):
        return "Error: arithmetic only"

    try:
        tree = ast.parse(expr, mode="eval")
        return str(_eval_ast(tree.body))  # type: ignore[attr-defined]
    except Exception as e:
        return f"Error: {e}"


calculator_agent = Agent(
    name="Calculator",
    instructions=(
        "Tu es une calculatrice précise. "
        "Quand on te donne un calcul, appelle l'outil eval_expression et renvoie uniquement le résultat numérique final. "
        "Pas de prose sauf si on te la demande."
    ),
    tools=[eval_expression],
    model_settings=ModelSettings(temperature=0),
)


# ==========================
# 4) Guardrail d'entrée : bloquer Tasha Yar
# ==========================
class YarGuardOutput(BaseModel):
    is_blocked: bool
    reasoning: str


guardrail_agent = Agent(
    name="Tasha Yar Guardrail",
    instructions=(
        "Tu es un guardrail. Détermine si l'utilisateur tente de parler de Tasha Yar (Star Trek: TNG).\n"
        "Retourne is_blocked=true si le texte mentionne Tasha Yar sous n'importe quelle forme "
        "(ex: 'Tasha Yar', 'Lt. Yar', 'Lieutenant Yar').\n"
        "Donne une raison en une phrase. Ne fournis que les champs du schéma de sortie."
    ),
    output_type=YarGuardOutput,
    model_settings=ModelSettings(temperature=0),
)


@input_guardrail
async def tasha_guardrail(
    ctx: RunContextWrapper[None],
    agent: Agent,
    input: Union[str, List[TResponseInputItem]],
) -> GuardrailFunctionOutput:
    result = await Runner.run(guardrail_agent, input, context=ctx.context)
    return GuardrailFunctionOutput(
        output_info=result.final_output.model_dump(),
        tripwire_triggered=bool(result.final_output.is_blocked),
    )


# ==========================
# 5) Agent Data : tools + handoff + guardrail
#    -> LAZY INIT pour éviter l'init vector store au moment de l'import
# ==========================
web_search = WebSearchTool()

_data_agent: Optional[Agent] = None


def get_data_agent() -> Agent:
    global _data_agent
    if _data_agent is not None:
        return _data_agent

    vs_id = ensure_vector_store_with_corpus(VECTOR_STORE_NAME, CORPUS_PATH)
    file_search = FileSearchTool(vector_store_ids=[vs_id], max_num_results=3)

    _data_agent = Agent(
        name="Lt. Cmdr. Data",
        instructions=(
            f"{RECOMMENDED_PROMPT_PREFIX}\n"
            "Tu es le Lieutenant Commander Data (Star Trek: TNG). Sois précis et concis (≤3 phrases).\n"
            "Utilise file_search pour les questions sur Data (RAG sur corpus), et web_search pour les faits récents sur le web.\n"
            "Si l'utilisateur demande un calcul arithmétique, FAIS UN HANDOFF vers l'agent Calculator."
        ),
        tools=[web_search, file_search],
        input_guardrails=[tasha_guardrail],
        handoffs=[calculator_agent],
        model_settings=ModelSettings(temperature=0),
    )
    return _data_agent


# ==========================
# 6) (Optionnel) Démo locale
# ==========================
async def main_demo_local():
    data_agent = get_data_agent()

    try:
        _ = await Runner.run(data_agent, "Tell me about your relationship with Tasha Yar.")
        print("ERROR: guardrail did not trip")
    except InputGuardrailTripwireTriggered:
        print("✅ Guardrail tripped as expected: Tasha Yar is off-limits.")

    ok = await Runner.run(data_agent, "Summarize Data's ethical subroutines in 2 sentences.")
    print("✅ Allowed prompt output:\n", ok.final_output)

    out = await Runner.run(data_agent, "Hello, Data. Please confirm your operational status.")
    print("\n[Agent] ", out.final_output)

    out = await Runner.run(data_agent, "Compute ((2*8)^2)/3")
    print("\n[Agent: math via calculator handoff] ", out.final_output)
    print("[Handled by agent]:", out.last_agent.name)

    out = await Runner.run(data_agent, "Do you experience emotions?")
    print("\n[Agent: file_search] ", out.final_output)
    print("[Handled by agent]:", out.last_agent.name)

    out = await Runner.run(
        data_agent,
        "Search the web for recent news about the James Webb Space Telescope and summarize briefly.",
    )
    print("\n[Agent: web_search] ", out.final_output)
    print("[Handled by agent]:", out.last_agent.name)


# ==========================
# 7) Intégration Bedrock AgentCore
# ==========================
app = BedrockAgentCoreApp()


@app.entrypoint
async def invoke(payload: dict):
    """
    Payload attendu (simple) :
      {"prompt": "Hello Data"}
    Réponse :
      {"result": "..."}
    """
    user_message = payload.get("prompt", "Data, reverse the main deflector array!")
    try:
        result = await Runner.run(get_data_agent(), user_message)
        output = result.final_output
    except InputGuardrailTripwireTriggered:
        output = "I'd really rather not talk about Tasha."

    return {"result": output}


# ==========================
# 8) Entrypoint
# ==========================
if __name__ == "__main__":
    if os.getenv("RUN_LOCAL_DEMO") == "1":
        import asyncio
        asyncio.run(main_demo_local())
    else:
        app.run()

import uuid

import boto3
import json

client = boto3.client('bedrock-agentcore', region_name='us-east-1')
payload = json.dumps({"prompt": "Data, tell me who created You"})

# session id valide (33+ chars)
session_id = str(uuid.uuid4()) + str(uuid.uuid4())


response = client.invoke_agent_runtime(
    agentRuntimeArn='arn:aws:bedrock-agentcore:us-east-1:851725292987:runtime/data_agent_agentcore-FWXUC891nI',
    runtimeSessionId=session_id,
    payload=payload,
    qualifier="DEFAULT" # This is Optional. When the field is not provided, Runtime will use DEFAULT endpoint
)
response_body = response['response'].read()
response_data = json.loads(response_body)
print("Agent Response:", response_data)