refactor(examples): compact agentcore run.py and test class

Co-Authored-By: mesutoezdil <versusfinem@gmail.com>

Author: mesutoezdil

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

@@ -1,17 +1,9 @@
 """Run a dice agent using AWS AgentCore with OTLP export.
 
-Shows how to send AgentCore traces to agentevals without using the
-agentevals SDK. The OTLP exporter points at the local receiver endpoint.
-
-AgentCore runs on Strands Agents with Amazon Bedrock models. This example
-sets up OTel via StrandsTelemetry and adds the OTLP exporter to the tracer
-provider before running the agent.
-
 Prerequisites:
     1. pip install -r requirements.txt
     2. agentevals serve --dev
-    3. AWS credentials configured (aws configure, env vars, or IAM role)
-    4. Amazon Nova Pro model access enabled in your AWS account
+    3. AWS credentials configured and Nova Pro model access enabled
 
 Usage:
     python examples/zero-code-examples/agentcore/run.py
@@ -36,11 +28,7 @@
 def roll_die(sides: int = 6) -> dict:
     """Roll a die with the given number of sides."""
     result = random.randint(1, sides)
-    return {
-        "sides": sides,
-        "result": result,
-        "message": f"Rolled a {sides}-sided die and got {result}",
-    }
+    return {"sides": sides, "result": result, "message": f"Rolled a {sides}-sided die and got {result}"}
 
 
 @tool
@@ -58,42 +46,27 @@ def main():
     if not os.getenv("AWS_DEFAULT_REGION"):
         print("AWS_DEFAULT_REGION not set.")
         return
-
     endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318")
     print(f"OTLP endpoint: {endpoint}")
-
     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)
     )
-
     agent = Agent(
         model=BedrockModel(model_id="us.amazon.nova-pro-v1:0"),
         tools=[roll_die, check_prime],
-        system_prompt=(
-            "You are a helpful assistant that can roll dice and check if numbers are prime. "
-            "Use roll_die with the right number of sides when asked. "
-            "Use check_prime when asked about prime numbers."
-        ),
+        system_prompt="You are a helpful assistant. Use roll_die when asked to roll dice. Use check_prime when asked about prime numbers.",
         name="dice_agent",
     )
 
-    test_queries = [
-        "Hi! Can you help me?",
-        "Roll a 20-sided die for me",
-        "Is the number you rolled prime?",
-    ]
-
-    for i, query in enumerate(test_queries, 1):
-        print(f"\n[{i}/{len(test_queries)}] User: {query}")
-        result = agent(query)
-        print(f"     Agent: {result}")
-
+    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}")
+        print(f"     Agent: {agent(query)}")
     print()
     telemetry.tracer_provider.force_flush()
     print("All traces flushed to OTLP receiver.")

tests/integration/test_live_agents.py

@@ -370,70 +370,41 @@ 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:
-    """Run the AWS AgentCore zero-code OTLP example and verify session grouping."""
-
     def test_session_created_spans_only(self, live_servers):
         main_port, otlp_http_port, mgr = live_servers
         session_name = "e2e-agentcore"
-
-        result = _run_agent(
-            "examples/zero-code-examples/agentcore/run.py",
-            otlp_http_port,
-            session_name,
-            extra_env={
-                "OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental",
-            },
-        )
+        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}"
-
         wait_for_session_complete_sync(mgr, session_name, timeout=60)
         session = mgr.sessions[session_name]
-
         assert session.is_complete
         assert session.source == "otlp"
         assert len(session.spans) > 0, "Expected spans from LLM calls"
 
     def test_invocations_extracted(self, live_servers):
         main_port, otlp_http_port, mgr = live_servers
         session_name = "e2e-agentcore-inv"
-
-        result = _run_agent(
-            "examples/zero-code-examples/agentcore/run.py",
-            otlp_http_port,
-            session_name,
-            extra_env={
-                "OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental",
-            },
-        )
+        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}"
-
         wait_for_session_complete_sync(mgr, session_name, timeout=60)
         session = mgr.sessions[session_name]
-
         assert len(session.invocations) > 0, "Expected extracted invocations"
 
     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(
-            "examples/zero-code-examples/agentcore/run.py",
-            otlp_http_port,
-            session_name,
-            extra_env={
-                "OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental",
-            },
-        )
+        result = _run_agent(_AGENTCORE_SCRIPT, otlp_http_port, session_name, extra_env=_AGENTCORE_ENV, timeout=60)
         assert result.returncode == 0
-
         wait_for_session_complete_sync(mgr, session_name, timeout=60)
-
         resp = httpx.get(f"http://127.0.0.1:{main_port}/api/streaming/sessions")
         assert resp.status_code == 200
-        session_ids = [s["sessionId"] for s in resp.json()["data"]]
-        assert session_name in session_ids
+        assert session_name in [s["sessionId"] for s in resp.json()["data"]]
 
 
 @_skip_no_openai