Python: Add samples syntax checking with pyright

python/.pre-commit-config.yaml

@@ -27,12 +27,6 @@ repos:
         name: Check Valid Python Samples
         types: ["python"]
         exclude: ^python/packages/lab/cookiecutter-agent-framework-lab/
-  - repo: https://github.com/nbQA-dev/nbQA
-    rev: 1.9.1
-    hooks:
-      - id: nbqa-check-ast
-        name: Check Valid Python Notebooks
-        types: ["jupyter"]
   - repo: https://github.com/asottile/pyupgrade
     rev: v3.20.0
     hooks:
@@ -47,6 +41,13 @@ repos:
         entry: uv --directory ./python run poe pre-commit-check
         language: system
         files: ^python/
+  - repo: https://github.com/PyCQA/bandit
+    rev: 1.8.5
+    hooks:
+      - id: bandit
+        name: Bandit Security Checks
+        args: ["-c", "python/pyproject.toml"]
+        additional_dependencies: ["bandit[toml]"]
   - repo: https://github.com/astral-sh/uv-pre-commit
     # uv version.
     rev: 0.7.18
@@ -56,10 +57,3 @@ repos:
         name: Update uv lockfile
         files: python/pyproject.toml
         args: [--project, python]
-  - repo: https://github.com/PyCQA/bandit
-    rev: 1.8.5
-    hooks:
-      - id: bandit
-        name: Bandit Security Checks
-        args: ["-c", "python/pyproject.toml"]
-        additional_dependencies: ["bandit[toml]"]

python/AGENTS.md

@@ -72,6 +72,12 @@ from agent_framework.azure import AzureOpenAIChatClient, AzureAIAgentClient
 
 When modifying samples, update associated README files in the same or parent folders.
 
+### Samples Syntax Checking
+
+Run `uv run poe samples-syntax` to check samples for syntax errors and missing imports from `agent_framework`. This uses a relaxed pyright configuration that validates imports without strict type checking.
+
+Some samples depend on external packages (e.g., `azure.ai.agentserver.agentframework`, `microsoft_agents`) that are not installed in the dev environment. These are excluded in `pyrightconfig.samples.json`. When adding or modifying these excluded samples, add them to the exclude list and manually verify they have no import errors from `agent_framework` packages by temporarily removing them from the exclude list and running the check.
+
 ## Package Documentation
 
 ### Core

python/packages/core/agent_framework/_types.py

@@ -6,6 +6,7 @@
 import json
 import re
 import sys
+from asyncio import iscoroutine
 from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable, Mapping, MutableMapping, Sequence
 from copy import deepcopy
 from typing import TYPE_CHECKING, Any, ClassVar, Final, Generic, Literal, NewType, cast, overload
@@ -2676,7 +2677,10 @@ async def _get_stream(self) -> AsyncIterable[TUpdate]:
             if hasattr(self._stream_source, "__aiter__"):
                 self._stream = self._stream_source  # type: ignore[assignment]
             else:
-                self._stream = await self._stream_source  # type: ignore[assignment]
+                if not iscoroutine(self._stream_source):
+                    self._stream = self._stream_source  # type: ignore[assignment]
+                else:
+                    self._stream = await self._stream_source  # type: ignore[assignment]
             if isinstance(self._stream, ResponseStream) and self._wrap_inner:
                 self._inner_stream = self._stream
                 return self._stream
@@ -2739,12 +2743,12 @@ async def get_final_response(self) -> TFinal:
         """
         if self._wrap_inner:
             if self._inner_stream is None:
-                if self._inner_stream_source is None:
-                    raise ValueError("No inner stream configured for this stream.")
-                if isinstance(self._inner_stream_source, ResponseStream):
-                    self._inner_stream = self._inner_stream_source
-                else:
-                    self._inner_stream = await self._inner_stream_source
+                # Use _get_stream() to resolve the awaitable - this properly handles
+                # the case where _stream_source and _inner_stream_source are the same
+                # coroutine (e.g., from from_awaitable), avoiding double-await errors.
+                await self._get_stream()
+            if self._inner_stream is None:
+                raise RuntimeError("Inner stream not available")
             if not self._finalized:
                 # Consume outer stream (which delegates to inner) if not already consumed
                 if not self._consumed:

python/pyproject.toml

@@ -225,8 +225,10 @@ test = "python run_tasks_in_packages_if_exists.py test"
 fmt = "python run_tasks_in_packages_if_exists.py fmt"
 format.ref = "fmt"
 lint = "python run_tasks_in_packages_if_exists.py lint"
+samples-lint = "ruff check samples --fix --exclude samples/autogen-migration,samples/semantic-kernel-migration --ignore E501,ASYNC,B901,TD002"
 pyright = "python run_tasks_in_packages_if_exists.py pyright"
 mypy = "python run_tasks_in_packages_if_exists.py mypy"
+samples-syntax = "pyright -p pyrightconfig.samples.json --warnings"
 typing = ["pyright", "mypy"]
 # cleaning
 clean-dist-packages = "python run_tasks_in_packages_if_exists.py clean-dist"
@@ -238,7 +240,7 @@ build-meta = "python -m flit build"
 build = ["build-packages", "build-meta"]
 publish = "uv publish"
 # combined checks
-check = ["fmt", "lint", "pyright", "mypy", "test", "markdown-code-lint"]
+check = ["fmt", "lint", "pyright", "mypy", "samples-lint", "samples-syntax", "test", "markdown-code-lint"]
 
 [tool.poe.tasks.all-tests-cov]
 cmd = """
@@ -323,7 +325,9 @@ sequence = [
     { ref = "fmt" },
     { ref = "lint" },
     { ref = "pre-commit-pyright ${files}" },
-    { ref = "pre-commit-markdown-code-lint ${files}" }
+    { ref = "pre-commit-markdown-code-lint ${files}" },
+    { ref = "samples-lint" },
+    { ref = "samples-syntax" }
 ]
 args = [{ name = "files", default = ".", positional = true, multiple = true }]
 

python/pyrightconfig.samples.json

@@ -0,0 +1,13 @@
+{
+    "include": ["samples"],
+    "exclude": [
+        "**/autogen/**",
+        "**/autogen-migration/**",
+        "**/semantic-kernel-migration/**",
+        "**/demos/**",
+        "**/agent_with_foundry_tracing.py"
+    ],
+    "typeCheckingMode": "off",
+    "reportMissingImports": "error",
+    "reportAttributeAccessIssue": "error"
+}

python/samples/autogen-migration/orchestrations/03_swarm.py

@@ -7,7 +7,7 @@
 
 import asyncio
 
-from agent_framework import WorkflowEvent
+from agent_framework import AgentResponseUpdate, WorkflowEvent
 from orderedmultidict import Any
 
 
@@ -99,7 +99,6 @@ async def run_autogen() -> None:
 async def run_agent_framework() -> None:
     """Agent Framework's HandoffBuilder for agent coordination."""
     from agent_framework import (
-        AgentResponseUpdate,
         WorkflowRunState,
     )
     from agent_framework.openai import OpenAIChatClient

python/samples/autogen-migration/single_agent/04_agent_as_tool.py

@@ -48,7 +48,7 @@ async def run_autogen() -> None:
 
 async def run_agent_framework() -> None:
     """Agent Framework's as_tool() for hierarchical agents with streaming."""
-    from agent_framework import FunctionCallContent, FunctionResultContent
+    from agent_framework import Content
     from agent_framework.openai import OpenAIChatClient
 
     client = OpenAIChatClient(model_id="gpt-4.1-mini")
@@ -78,7 +78,7 @@ async def run_agent_framework() -> None:
     print("[Agent Framework]")
 
     # Track accumulated function calls (they stream in incrementally)
-    accumulated_calls: dict[str, FunctionCallContent] = {}
+    accumulated_calls: dict[str, Content] = {}
 
     async for chunk in coordinator.run("Create a tagline for a coffee shop", stream=True):
         # Stream text tokens
@@ -88,7 +88,7 @@ async def run_agent_framework() -> None:
         # Process streaming function calls and results
         if chunk.contents:
             for content in chunk.contents:
-                if isinstance(content, FunctionCallContent):
+                if content.type == "function_call":
                     # Accumulate function call content as it streams in
                     call_id = content.call_id
                     if call_id in accumulated_calls:
@@ -105,7 +105,7 @@ async def run_agent_framework() -> None:
                     current_args = accumulated_calls[call_id].arguments
                     print(f"  Arguments: {current_args}", flush=True)
 
-                elif isinstance(content, FunctionResultContent):
+                elif content.type == "function_result":
                     # Tool result - shows writer's response
                     result_text = content.result if isinstance(content.result, str) else str(content.result)
                     if result_text.strip():

python/samples/concepts/response_stream.py

@@ -154,11 +154,11 @@ async def generate_updates() -> AsyncIterable[ChatResponseUpdate]:
         words = ["Hello", " ", "from", " ", "the", " ", "streaming", " ", "response", "!"]
         for word in words:
             await asyncio.sleep(0.05)  # Simulate network delay
-            yield ChatResponseUpdate(contents=[Content.from_text(word)], role=Role.ASSISTANT)
+            yield ChatResponseUpdate(contents=[Content.from_text(word)], role="assistant")
 
     def combine_updates(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
         """Finalizer that combines all updates into a single response."""
-        return ChatResponse.from_chat_response_updates(updates)
+        return ChatResponse.from_updates(updates)
 
     stream = ResponseStream(generate_updates(), finalizer=combine_updates)
 
@@ -237,7 +237,7 @@ async def cleanup_hook() -> None:
     )
 
     print("Starting iteration (cleanup happens after):")
-    async for update in stream4:
+    async for _update in stream4:
         pass  # Just consume the stream
     print(f"Cleanup was performed: {cleanup_performed['value']}")
 

python/samples/demos/chatkit-integration/app.py

@@ -18,7 +18,7 @@
 import uvicorn
 
 # Agent Framework imports
-from agent_framework import AgentResponseUpdate, ChatAgent, ChatMessage, FunctionResultContent, Role, tool
+from agent_framework import AgentResponseUpdate, ChatAgent, ChatMessage, tool
 from agent_framework.azure import AzureOpenAIChatClient
 
 # Agent Framework ChatKit integration
@@ -281,7 +281,7 @@ async def _update_thread_title(
 
             title_prompt = [
                 ChatMessage(
-                    role=Role.USER,
+                    role="user",
                     text=(
                         f"Generate a very short, concise title (max 40 characters) for a conversation "
                         f"that starts with:\n\n{conversation_context}\n\n"
@@ -332,7 +332,6 @@ async def respond(
         runs the agent, converts the response back to ChatKit events using stream_agent_response,
         and creates interactive weather widgets when weather data is queried.
         """
-        from agent_framework import FunctionResultContent
 
         if input_user_message is None:
             logger.debug("Received None user message, skipping")
@@ -375,7 +374,7 @@ async def intercept_stream() -> AsyncIterator[AgentResponseUpdate]:
                     # Check for function results in the update
                     if update.contents:
                         for content in update.contents:
-                            if isinstance(content, FunctionResultContent):
+                            if content.type == "function_result":
                                 result = content.result
 
                                 # Check if it's a WeatherResponse (string subclass with weather_data attribute)
@@ -458,7 +457,7 @@ async def action(
             weather_data: WeatherData | None = None
 
             # Create an agent message asking about the weather
-            agent_messages = [ChatMessage(role=Role.USER, text=f"What's the weather in {city_label}?")]
+            agent_messages = [ChatMessage(role="user", text=f"What's the weather in {city_label}?")]
 
             logger.debug(f"Processing weather query: {agent_messages[0].text}")
 
@@ -472,7 +471,7 @@ async def intercept_stream() -> AsyncIterator[AgentResponseUpdate]:
                     # Check for function results in the update
                     if update.contents:
                         for content in update.contents:
-                            if isinstance(content, FunctionResultContent):
+                            if content.type == "function_result":
                                 result = content.result
 
                                 # Check if it's a WeatherResponse (string subclass with weather_data attribute)
@@ -563,7 +562,7 @@ async def chatkit_endpoint(request: Request):
 
 
 @app.post("/upload/{attachment_id}")
-async def upload_file(attachment_id: str, file: UploadFile = File(...)):
+async def upload_file(attachment_id: str, file: Annotated[UploadFile, File()]):
     """Handle file upload for two-phase upload.
 
     The client POSTs the file bytes here after creating the attachment
@@ -585,7 +584,7 @@ async def upload_file(attachment_id: str, file: UploadFile = File(...)):
         attachment = await data_store.load_attachment(attachment_id, {"user_id": DEFAULT_USER_ID})
 
         # Clear the upload_url since upload is complete
-        attachment.upload_url = None
+        attachment.upload_url = None  # type: ignore[union-attr]
 
         # Save the updated attachment back to the store
         await data_store.save_attachment(attachment, {"user_id": DEFAULT_USER_ID})

python/samples/demos/hosted_agents/agents_in_workflow/main.py

@@ -1,7 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from agent_framework import ConcurrentBuilder
 from agent_framework.azure import AzureOpenAIChatClient
+from agent_framework_orchestrations import ConcurrentBuilder
 from azure.ai.agentserver.agentframework import from_agent_framework
 from azure.identity import DefaultAzureCredential  # pyright: ignore[reportUnknownVariableType]