refactor(examples): use BedrockAgentCoreApp for AgentCore example

Rewrites the AgentCore zero-code example to run as a proper AgentCore
server using BedrockAgentCoreApp from the bedrock-agentcore SDK. The
handler is now an async generator registered with @app.entrypoint and
the server starts with app.run(), matching how AgentCore expects agents
to run.

Updates the e2e tests to use subprocess.Popen instead of subprocess.run,
wait for /ping health check, then POST to /invocations. Tests pass
without AWS credentials because Strands exports OTLP spans even when
BedrockModel raises NoCredentialsError.

Author: mesutoezdil

examples/README.md

@@ -222,7 +222,10 @@ python examples/zero-code-examples/ollama/run.py
 python examples/zero-code-examples/strands/run.py
 python examples/zero-code-examples/adk/run.py
 python examples/zero-code-examples/pydantic-ai/run.py
-python examples/zero-code-examples/agentcore/run.py
+
+# AgentCore starts a server (AWS credentials required):
+python examples/zero-code-examples/agentcore/run.py &
+curl http://localhost:8080/invocations -d '{"prompt": "Roll a 20-sided die for me"}'
 
 # SDK examples:
 python examples/sdk_example/context_manager_example.py

examples/zero-code-examples/agentcore/requirements.txt

@@ -1,3 +1,4 @@
+bedrock-agentcore>=1.8.0
 strands-agents>=1.35.0
 boto3>=1.38.0
 opentelemetry-sdk>=1.36.0

examples/zero-code-examples/agentcore/run.py

@@ -1,10 +1,26 @@
-"""AWS AgentCore zero-code OTLP example. Needs AWS credentials and Nova Pro model access.
-Run: python examples/zero-code-examples/agentcore/run.py
+"""AWS AgentCore zero-code OTLP example -- no agentevals SDK.
+
+Demonstrates zero-code integration: wraps a Strands agent with BedrockAgentCoreApp
+so it runs on the AgentCore runtime. The only agentevals setup is a standard
+OTLPSpanExporter pointing at the receiver.
+
+Prerequisites:
+    1. pip install -r examples/zero-code-examples/agentcore/requirements.txt
+    2. agentevals serve --dev
+    3. export AWS_DEFAULT_REGION=us-east-1  (Nova Pro model access required)
+
+Usage (direct):
+    python examples/zero-code-examples/agentcore/run.py
+    curl http://localhost:8080/invocations -d '{"prompt": "Roll a 20-sided die"}'
+
+Usage (via AgentCore CLI, installed with npm install -g @aws/agentcore):
+    agentcore dev
 """
 
 import os
 import random
 
+from bedrock_agentcore import BedrockAgentCoreApp
 from dotenv import load_dotenv
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
@@ -14,6 +30,13 @@
 
 load_dotenv(override=True)
 os.environ.setdefault("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
+os.environ.setdefault("OTEL_RESOURCE_ATTRIBUTES",
+    "agentevals.eval_set_id=agentcore_eval,agentevals.session_name=agentcore-zero-code")
+
+_telemetry = StrandsTelemetry()
+_telemetry.tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(), schedule_delay_millis=1000))
+
+app = BedrockAgentCoreApp()
 
 
 @tool
@@ -29,26 +52,17 @@ def check_prime(n: int) -> bool:
     return n >= 2 and all(n % i for i in range(2, int(n**0.5) + 1))
 
 
-def main():
-    if not os.getenv("AWS_DEFAULT_REGION"):
-        print("AWS_DEFAULT_REGION not set.")
-        return
-    os.environ.setdefault("OTEL_RESOURCE_ATTRIBUTES",
-        "agentevals.eval_set_id=agentcore_eval,agentevals.session_name=agentcore-zero-code")
-    telemetry = StrandsTelemetry()
-    telemetry.tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(), schedule_delay_millis=1000))
+@app.entrypoint
+async def handler(payload):
+    prompt = payload.get("prompt", "Hello!")
     agent = Agent(
         model=BedrockModel(model_id="us.amazon.nova-pro-v1:0"),
         tools=[roll_die, check_prime],
-        system_prompt="You are a helpful assistant. Use roll_die when asked to roll dice. Use check_prime when asked about prime numbers.",
+        system_prompt="Use roll_die when asked to roll dice. Use check_prime when asked about prime numbers.",
         name="dice_agent",
     )
-    queries = ["Hi! Can you help me?", "Roll a 20-sided die for me", "Is the number you rolled prime?"]
-    for i, query in enumerate(queries, 1):
-        print(f"\n[{i}/{len(queries)}] User: {query}\n     Agent: {agent(query)}")
-    telemetry.tracer_provider.force_flush()
-    print("All traces flushed.")
+    async for event in agent.stream_async(prompt):
+        yield event
 
 
-if __name__ == "__main__":
-    main()
+app.run()

tests/integration/test_live_agents.py

@@ -15,9 +15,11 @@
 
 from __future__ import annotations
 
+import contextlib
 import os
 import subprocess
 import sys
+import time
 
 import httpx
 import pytest
@@ -38,9 +40,6 @@
     reason="GOOGLE_API_KEY not set",
 )
 
-_skip_no_aws = pytest.mark.skipif(not os.environ.get("AWS_DEFAULT_REGION"), reason="AWS_DEFAULT_REGION not set")
-
-
 def _run_agent(
     script: str,
     otlp_http_port: int,
@@ -66,6 +65,46 @@ def _run_agent(
     )
 
 
+_AGENTCORE_ENV = {"OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental"}
+_AGENTCORE_SCRIPT = "examples/zero-code-examples/agentcore/run.py"
+
+
+@contextlib.contextmanager
+def _agentcore_server(otlp_http_port: int, session_name: str, extra_env: dict | None = None):
+    env = {
+        **os.environ,
+        "OTEL_EXPORTER_OTLP_ENDPOINT": f"http://127.0.0.1:{otlp_http_port}",
+        "OTEL_RESOURCE_ATTRIBUTES": f"agentevals.eval_set_id=e2e-test,agentevals.session_name={session_name}",
+        **(extra_env or {}),
+    }
+    proc = subprocess.Popen(
+        [sys.executable, os.path.join(REPO_ROOT, _AGENTCORE_SCRIPT)],
+        env=env,
+        cwd=REPO_ROOT,
+    )
+    try:
+        for _ in range(20):
+            if proc.poll() is not None:
+                raise RuntimeError(f"agentcore server exited early (code {proc.returncode})")
+            try:
+                httpx.get("http://127.0.0.1:8080/ping", timeout=1)
+                break
+            except Exception:
+                time.sleep(0.5)
+        else:
+            proc.kill()
+            raise RuntimeError("agentcore server did not start within 10s")
+        yield proc
+    finally:
+        time.sleep(2)
+        proc.terminate()
+        try:
+            proc.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            proc.wait()
+
+
 @_skip_no_openai
 class TestLangchainZeroCode:
     """Run the LangChain zero-code OTLP example and verify session grouping."""
@@ -367,34 +406,30 @@ def test_session_visible_via_api(self, live_servers):
         assert session_name in session_ids
 
 
-_AGENTCORE_ENV = {"OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental"}
-_AGENTCORE_SCRIPT = "examples/zero-code-examples/agentcore/run.py"
-
 
-@_skip_no_aws
 class TestAgentCoreZeroCode:
     def test_session_created_spans_only(self, live_servers):
         main_port, otlp_http_port, mgr = live_servers
         session_name = "e2e-agentcore"
-        result = _run_agent(_AGENTCORE_SCRIPT, otlp_http_port, session_name, extra_env=_AGENTCORE_ENV, timeout=60)
-        assert result.returncode == 0, f"Agent failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
+        with _agentcore_server(otlp_http_port, session_name, extra_env=_AGENTCORE_ENV):
+            httpx.post("http://127.0.0.1:8080/invocations", json={"prompt": "Roll a 20-sided die"}, timeout=60)
         wait_for_session_complete_sync(mgr, session_name, timeout=60)
         s = mgr.sessions[session_name]
         assert s.is_complete and s.source == "otlp" and len(s.spans) > 0
 
     def test_invocations_extracted(self, live_servers):
         main_port, otlp_http_port, mgr = live_servers
         session_name = "e2e-agentcore-inv"
-        result = _run_agent(_AGENTCORE_SCRIPT, otlp_http_port, session_name, extra_env=_AGENTCORE_ENV, timeout=60)
-        assert result.returncode == 0, f"Agent failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
+        with _agentcore_server(otlp_http_port, session_name, extra_env=_AGENTCORE_ENV):
+            httpx.post("http://127.0.0.1:8080/invocations", json={"prompt": "Is 17 prime?"}, timeout=60)
         wait_for_session_complete_sync(mgr, session_name, timeout=60)
         assert len(mgr.sessions[session_name].invocations) > 0
 
     def test_session_visible_via_api(self, live_servers):
         main_port, otlp_http_port, mgr = live_servers
         session_name = "e2e-agentcore-api"
-        result = _run_agent(_AGENTCORE_SCRIPT, otlp_http_port, session_name, extra_env=_AGENTCORE_ENV, timeout=60)
-        assert result.returncode == 0
+        with _agentcore_server(otlp_http_port, session_name, extra_env=_AGENTCORE_ENV):
+            httpx.post("http://127.0.0.1:8080/invocations", json={"prompt": "Hello!"}, timeout=60)
         wait_for_session_complete_sync(mgr, session_name, timeout=60)
         data = httpx.get(f"http://127.0.0.1:{main_port}/api/streaming/sessions").json()["data"]
         assert session_name in [s["sessionId"] for s in data]