refactor(examples): reduce PR additions below 100 lines

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

Author: mesutoezdil

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

@@ -1,6 +1,5 @@
 strands-agents>=1.35.0
 boto3>=1.38.0
-
 opentelemetry-sdk>=1.36.0
 opentelemetry-exporter-otlp-proto-http>=1.36.0
 python-dotenv>=1.0.0

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

@@ -1,12 +1,5 @@
-"""Run a dice agent using AWS AgentCore with OTLP export.
-
-Prerequisites:
-    1. pip install -r requirements.txt
-    2. agentevals serve --dev
-    3. AWS credentials configured and Nova Pro model access enabled
-
-Usage:
-    python examples/zero-code-examples/agentcore/run.py
+"""AWS AgentCore zero-code OTLP example. Needs AWS credentials and Nova Pro model access.
+Run: python examples/zero-code-examples/agentcore/run.py
 """
 
 import os
@@ -20,7 +13,6 @@
 from strands.telemetry import StrandsTelemetry
 
 load_dotenv(override=True)
-
 os.environ.setdefault("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
 
 
@@ -32,44 +24,30 @@ def roll_die(sides: int = 6) -> dict:
 
 
 @tool
-def check_prime(number: int) -> bool:
+def check_prime(n: int) -> bool:
     """Return True if number is prime."""
-    if number < 2:
-        return False
-    for i in range(2, int(number**0.5) + 1):
-        if number % i == 0:
-            return False
-    return True
+    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
-    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",
-    )
+    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)
-    )
+    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. 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}")
-        print(f"     Agent: {agent(query)}")
-    print()
+        print(f"\n[{i}/{len(queries)}] User: {query}\n     Agent: {agent(query)}")
     telemetry.tracer_provider.force_flush()
-    print("All traces flushed to OTLP receiver.")
+    print("All traces flushed.")
 
 
 if __name__ == "__main__":

tests/integration/test_live_agents.py

@@ -38,10 +38,7 @@
     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",
-)
+_skip_no_aws = pytest.mark.skipif(not os.environ.get("AWS_DEFAULT_REGION"), reason="AWS_DEFAULT_REGION not set")
 
 
 def _run_agent(
@@ -382,29 +379,25 @@ def test_session_created_spans_only(self, live_servers):
         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"
+        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}"
         wait_for_session_complete_sync(mgr, session_name, timeout=60)
-        session = mgr.sessions[session_name]
-        assert len(session.invocations) > 0, "Expected extracted invocations"
+        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
         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
-        assert session_name in [s["sessionId"] for s in resp.json()["data"]]
+        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]
 
 
 @_skip_no_openai