Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Tao Chen
|
db4b4c2736 | ||
|
Tao Chen
|
ad654b523a | ||
|
Tao Chen
|
172c0a9507 |
+1
@@ -0,0 +1 @@
|
||||
../../../.github/skills/pull-requests
|
||||
-116
@@ -1,116 +0,0 @@
|
||||
---
|
||||
name: pull-requests
|
||||
description: >
|
||||
Guidance for creating pull requests and handling PR review comments in the
|
||||
Agent Framework repository. Use this when writing a PR description (filling out
|
||||
the PR template) or when responding to and resolving review comments on an
|
||||
existing PR.
|
||||
---
|
||||
|
||||
# Pull Request Workflow
|
||||
|
||||
This skill covers two tasks: (1) writing a high-quality PR description, and
|
||||
(2) handling review comments on an existing PR.
|
||||
|
||||
## 1. Writing the PR description
|
||||
|
||||
Always follow the repository PR template at
|
||||
[`.github/pull_request_template.md`](../../../../.github/pull_request_template.md). Keep its
|
||||
exact structure and headings. Fill every section:
|
||||
|
||||
### `### Motivation & Context`
|
||||
Explain *why* the change is needed: the problem it solves and the scenario it
|
||||
contributes to. Describe the net change relative to `main` — this is implied, so
|
||||
do **not** spell out "vs main" explicitly.
|
||||
|
||||
### `### Description & Review Guide`
|
||||
Describe the changes, the overall approach, and the design. Answer the three
|
||||
prompts:
|
||||
- **What are the major changes?**
|
||||
- **What is the impact of these changes?**
|
||||
- **What do you want reviewers to focus on?** — This item is for **human
|
||||
reviewers only**. Automated/AI reviewers must ignore it and review the entire
|
||||
change rather than narrowing scope to it.
|
||||
|
||||
### `### Related Issue`
|
||||
Link the issue the PR fixes using a GitHub closing keyword (`Fixes #123` /
|
||||
`Closes #123`) so it closes automatically on merge. A PR with no linked issue may
|
||||
be closed regardless of how valid the change is. Before opening, confirm there is
|
||||
no other open PR for the same issue; if there is, explain how this PR differs.
|
||||
|
||||
### `### Contribution Checklist`
|
||||
Check every item that applies. For the breaking-change item:
|
||||
- Leave **"This is not a breaking change."** checked for the common case.
|
||||
- If the change **is** breaking, add the `breaking change` label **or** put
|
||||
`[BREAKING]` in the title prefix, before or after a language prefix such as
|
||||
`Python:` or `.NET:` — workflows keep the label and the title prefix in sync
|
||||
automatically (see `.github/workflows/label-title-prefix.yml` and
|
||||
`.github/workflows/label-pr.yml`).
|
||||
|
||||
### Do not
|
||||
- Do **not** add ad-hoc sections such as "Validation" or "Tests run"; CI/CD and
|
||||
the checklist already cover validation status.
|
||||
- Do **not** remove or reorder the template's headings.
|
||||
|
||||
### Creating the PR
|
||||
Open new PRs as **drafts** until they are ready for review. Example:
|
||||
|
||||
```bash
|
||||
gh pr create --repo microsoft/agent-framework --base main \
|
||||
--head <your-fork-owner>:<branch> --draft \
|
||||
--title "<concise title>" --body "<body following the template>"
|
||||
```
|
||||
|
||||
## 2. Handling review comments
|
||||
|
||||
When a PR receives review comments, follow this sequence — **do not start editing
|
||||
code before the user has reviewed the plan**:
|
||||
|
||||
1. **Review the comments.** Read every review comment and thread on the PR,
|
||||
including inline code comments and general review summaries.
|
||||
2. **Make a plan.** Produce a concrete plan describing how each comment will be
|
||||
addressed (or why it should not be, with reasoning).
|
||||
3. **Let the user review the plan.** Present the plan and wait for the user's
|
||||
approval or adjustments before implementing anything.
|
||||
4. **Implement.** Make the agreed changes.
|
||||
5. **Reply to every comment.** Add a reply to **all** comments explaining how it
|
||||
was addressed (or the agreed outcome) — leave none unanswered.
|
||||
6. **Resolve resolved threads.** Mark a review thread as resolved only when the
|
||||
comment has actually been addressed.
|
||||
|
||||
### Useful commands
|
||||
|
||||
List review comments and threads:
|
||||
|
||||
```bash
|
||||
# Inline review comments
|
||||
gh api repos/{owner}/{repo}/pulls/{pr}/comments
|
||||
|
||||
# Review threads with resolution state (GraphQL)
|
||||
gh api graphql -f query='
|
||||
query($owner:String!,$repo:String!,$pr:Int!){
|
||||
repository(owner:$owner,name:$repo){
|
||||
pullRequest(number:$pr){
|
||||
reviewThreads(first:100){
|
||||
nodes{ id isResolved comments(first:50){ nodes{ id body author{login} } } }
|
||||
}
|
||||
}
|
||||
}
|
||||
}' -F owner={owner} -F repo={repo} -F pr={pr}
|
||||
```
|
||||
|
||||
Reply to an inline review comment:
|
||||
|
||||
```bash
|
||||
gh api repos/{owner}/{repo}/pulls/{pr}/comments/{comment_id}/replies \
|
||||
-f body="Addressed in <commit>: <explanation>"
|
||||
```
|
||||
|
||||
Resolve a review thread (needs the thread node id from the GraphQL query above):
|
||||
|
||||
```bash
|
||||
gh api graphql -f query='
|
||||
mutation($threadId:ID!){
|
||||
resolveReviewThread(input:{threadId:$threadId}){ thread{ isResolved } }
|
||||
}' -F threadId={thread_id}
|
||||
```
|
||||
@@ -3,7 +3,7 @@
|
||||
<!-- Central version prefix - applies to all nuget packages. -->
|
||||
<VersionPrefix>1.11.0</VersionPrefix>
|
||||
<RCNumber>1</RCNumber>
|
||||
<DateSuffix>260623</DateSuffix>
|
||||
<DateSuffix>260622</DateSuffix>
|
||||
<PackageVersion Condition="'$(IsReleaseCandidate)' == 'true'">$(VersionPrefix)-rc$(RCNumber)</PackageVersion>
|
||||
<PackageVersion Condition="'$(IsReleaseCandidate)' != 'true' AND '$(VersionSuffix)' != ''">$(VersionPrefix)-$(VersionSuffix).$(DateSuffix).1</PackageVersion>
|
||||
<PackageVersion Condition="'$(IsReleaseCandidate)' != 'true' AND '$(VersionSuffix)' == ''">$(VersionPrefix)-preview.$(DateSuffix).1</PackageVersion>
|
||||
|
||||
+1
@@ -0,0 +1 @@
|
||||
../../../.github/skills/pull-requests
|
||||
-116
@@ -1,116 +0,0 @@
|
||||
---
|
||||
name: pull-requests
|
||||
description: >
|
||||
Guidance for creating pull requests and handling PR review comments in the
|
||||
Agent Framework repository. Use this when writing a PR description (filling out
|
||||
the PR template) or when responding to and resolving review comments on an
|
||||
existing PR.
|
||||
---
|
||||
|
||||
# Pull Request Workflow
|
||||
|
||||
This skill covers two tasks: (1) writing a high-quality PR description, and
|
||||
(2) handling review comments on an existing PR.
|
||||
|
||||
## 1. Writing the PR description
|
||||
|
||||
Always follow the repository PR template at
|
||||
[`.github/pull_request_template.md`](../../../../.github/pull_request_template.md). Keep its
|
||||
exact structure and headings. Fill every section:
|
||||
|
||||
### `### Motivation & Context`
|
||||
Explain *why* the change is needed: the problem it solves and the scenario it
|
||||
contributes to. Describe the net change relative to `main` — this is implied, so
|
||||
do **not** spell out "vs main" explicitly.
|
||||
|
||||
### `### Description & Review Guide`
|
||||
Describe the changes, the overall approach, and the design. Answer the three
|
||||
prompts:
|
||||
- **What are the major changes?**
|
||||
- **What is the impact of these changes?**
|
||||
- **What do you want reviewers to focus on?** — This item is for **human
|
||||
reviewers only**. Automated/AI reviewers must ignore it and review the entire
|
||||
change rather than narrowing scope to it.
|
||||
|
||||
### `### Related Issue`
|
||||
Link the issue the PR fixes using a GitHub closing keyword (`Fixes #123` /
|
||||
`Closes #123`) so it closes automatically on merge. A PR with no linked issue may
|
||||
be closed regardless of how valid the change is. Before opening, confirm there is
|
||||
no other open PR for the same issue; if there is, explain how this PR differs.
|
||||
|
||||
### `### Contribution Checklist`
|
||||
Check every item that applies. For the breaking-change item:
|
||||
- Leave **"This is not a breaking change."** checked for the common case.
|
||||
- If the change **is** breaking, add the `breaking change` label **or** put
|
||||
`[BREAKING]` in the title prefix, before or after a language prefix such as
|
||||
`Python:` or `.NET:` — workflows keep the label and the title prefix in sync
|
||||
automatically (see `.github/workflows/label-title-prefix.yml` and
|
||||
`.github/workflows/label-pr.yml`).
|
||||
|
||||
### Do not
|
||||
- Do **not** add ad-hoc sections such as "Validation" or "Tests run"; CI/CD and
|
||||
the checklist already cover validation status.
|
||||
- Do **not** remove or reorder the template's headings.
|
||||
|
||||
### Creating the PR
|
||||
Open new PRs as **drafts** until they are ready for review. Example:
|
||||
|
||||
```bash
|
||||
gh pr create --repo microsoft/agent-framework --base main \
|
||||
--head <your-fork-owner>:<branch> --draft \
|
||||
--title "<concise title>" --body "<body following the template>"
|
||||
```
|
||||
|
||||
## 2. Handling review comments
|
||||
|
||||
When a PR receives review comments, follow this sequence — **do not start editing
|
||||
code before the user has reviewed the plan**:
|
||||
|
||||
1. **Review the comments.** Read every review comment and thread on the PR,
|
||||
including inline code comments and general review summaries.
|
||||
2. **Make a plan.** Produce a concrete plan describing how each comment will be
|
||||
addressed (or why it should not be, with reasoning).
|
||||
3. **Let the user review the plan.** Present the plan and wait for the user's
|
||||
approval or adjustments before implementing anything.
|
||||
4. **Implement.** Make the agreed changes.
|
||||
5. **Reply to every comment.** Add a reply to **all** comments explaining how it
|
||||
was addressed (or the agreed outcome) — leave none unanswered.
|
||||
6. **Resolve resolved threads.** Mark a review thread as resolved only when the
|
||||
comment has actually been addressed.
|
||||
|
||||
### Useful commands
|
||||
|
||||
List review comments and threads:
|
||||
|
||||
```bash
|
||||
# Inline review comments
|
||||
gh api repos/{owner}/{repo}/pulls/{pr}/comments
|
||||
|
||||
# Review threads with resolution state (GraphQL)
|
||||
gh api graphql -f query='
|
||||
query($owner:String!,$repo:String!,$pr:Int!){
|
||||
repository(owner:$owner,name:$repo){
|
||||
pullRequest(number:$pr){
|
||||
reviewThreads(first:100){
|
||||
nodes{ id isResolved comments(first:50){ nodes{ id body author{login} } } }
|
||||
}
|
||||
}
|
||||
}
|
||||
}' -F owner={owner} -F repo={repo} -F pr={pr}
|
||||
```
|
||||
|
||||
Reply to an inline review comment:
|
||||
|
||||
```bash
|
||||
gh api repos/{owner}/{repo}/pulls/{pr}/comments/{comment_id}/replies \
|
||||
-f body="Addressed in <commit>: <explanation>"
|
||||
```
|
||||
|
||||
Resolve a review thread (needs the thread node id from the GraphQL query above):
|
||||
|
||||
```bash
|
||||
gh api graphql -f query='
|
||||
mutation($threadId:ID!){
|
||||
resolveReviewThread(input:{threadId:$threadId}){ thread{ isResolved } }
|
||||
}' -F threadId={thread_id}
|
||||
```
|
||||
@@ -34,8 +34,6 @@ Status is grouped into these buckets:
|
||||
| `agent-framework-foundry-local` | `python/packages/foundry_local` | `beta` |
|
||||
| `agent-framework-gemini` | `python/packages/gemini` | `alpha` |
|
||||
| `agent-framework-github-copilot` | `python/packages/github_copilot` | `rc` |
|
||||
| `agent-framework-hosting` | `python/packages/hosting` | `alpha` |
|
||||
| `agent-framework-hosting-responses` | `python/packages/hosting-responses` | `alpha` |
|
||||
| `agent-framework-hyperlight` | `python/packages/hyperlight` | `beta` |
|
||||
| `agent-framework-lab` | `python/packages/lab` | `beta` |
|
||||
| `agent-framework-mem0` | `python/packages/mem0` | `beta` |
|
||||
|
||||
@@ -80,9 +80,8 @@ agent_framework/
|
||||
|
||||
- **`MCPTool`** - Base wrapper that owns the MCP `ClientSession` and exposes the remote server's tools as `FunctionTool`s.
|
||||
- **`MCPStdioTool`** / **`MCPStreamableHTTPTool`** / **`MCPWebsocketTool`** - Transport-specific subclasses.
|
||||
- **Argument allowlist (`_prepare_call_kwargs`)** - Before each `tools/call`, kwargs are filtered to an **allowlist** built from the tool's declared parameters (`inputSchema.properties`) plus any user-configured extras. Framework runtime kwargs injected through the function-invocation pipeline (e.g. `thread`, `conversation_id`, `chat_options`, `options`, `response_format`) are stripped by default rather than forwarded. A tool that declares no usable `properties` (including schemas with `additionalProperties: true`) forwards only the configured extras. The `_MCP_FRAMEWORK_DENYLIST` is a safety net for framework-named params a server *declares* in its schema (those are dropped); names explicitly opted in via `additional_tool_argument_names` always win. The reserved `_meta` key is never forwarded as an argument; trusted caller/runtime `_meta` is validated as MCP request metadata, model-supplied `_meta` is discarded in generated MCP functions, and metadata precedence is caller/runtime < OpenTelemetry < tools/list metadata.
|
||||
- **`allowed_tools`** (constructor arg on all `MCPTool` subclasses) - Restricts exposed MCP tools by raw remote MCP tool identity. Prefixed local names remain accepted only when the raw remote name already matches its normalized form; normalized/local aliases do not authorize a different raw remote name. If multiple raw remote tool names map to the same local function name, tool loading raises `ToolExecutionException` instead of first-one-wins shadowing.
|
||||
- **`additional_tool_argument_names`** (constructor arg on all `MCPTool` subclasses) - Opt extra argument names back into the allowlist. Accepts a `Sequence[str]` (applied to every tool) or a `Mapping[str, Sequence[str]]` keyed by **remote tool name**, where the reserved key `"*"` denotes global extras. It is configured only in user code at construction; there is **no per-call/runtime override**, so a model-issued tool call cannot change which names pass through. To use a server that accepts `additionalProperties: true`, list the extra names here and then either (1) manually extend that tool's `inputSchema` (via the `.functions` list after connecting) so the model is prompted to supply them, or (2) supply the values yourself via `function_invocation_kwargs`. If a normal forwarded argument name is supplied by both the model and `function_invocation_kwargs`, the model-supplied value wins; `_meta` is the exception and only trusted runtime/caller metadata is used.
|
||||
- **Argument allowlist (`_prepare_call_kwargs`)** - Before each `tools/call`, kwargs are filtered to an **allowlist** built from the tool's declared parameters (`inputSchema.properties`) plus any user-configured extras. Framework runtime kwargs injected through the function-invocation pipeline (e.g. `thread`, `conversation_id`, `chat_options`, `options`, `response_format`) are stripped by default rather than forwarded. A tool that declares no usable `properties` (including schemas with `additionalProperties: true`) forwards only the configured extras. The `_MCP_FRAMEWORK_DENYLIST` is a safety net for framework-named params a server *declares* in its schema (those are dropped); names explicitly opted in via `additional_tool_argument_names` always win. The reserved `_meta` key is extracted as MCP request metadata, never forwarded as an argument.
|
||||
- **`additional_tool_argument_names`** (constructor arg on all `MCPTool` subclasses) - Opt extra argument names back into the allowlist. Accepts a `Sequence[str]` (applied to every tool) or a `Mapping[str, Sequence[str]]` keyed by **remote tool name**, where the reserved key `"*"` denotes global extras. It is configured only in user code at construction; there is **no per-call/runtime override**, so a model-issued tool call cannot change which names pass through. To use a server that accepts `additionalProperties: true`, list the extra names here and then either (1) manually extend that tool's `inputSchema` (via the `.functions` list after connecting) so the model is prompted to supply them, or (2) supply the values yourself via `function_invocation_kwargs`. If a name is supplied by both the model and `function_invocation_kwargs`, the model-supplied value wins.
|
||||
- **Sampling guardrails** (`sampling_callback`) - Passing `client=` advertises `SamplingCapability` so the server can send `sampling/createMessage`. Because remote servers are untrusted (confused-deputy risk), the default `sampling_callback` is **deny-by-default** and applies, in order: a per-session rate limit (`sampling_max_requests`, default `_DEFAULT_SAMPLING_MAX_REQUESTS`), an approval gate (`sampling_approval_callback`), and a `maxTokens` cap (`sampling_max_tokens`, default `_DEFAULT_SAMPLING_MAX_TOKENS`). The approval callback (constructor arg on all subclasses; exported type alias `SamplingApprovalCallback`) receives the raw `CreateMessageRequestParams`, may be sync or async, and must return truthy to approve. When it is `None` (the default) every sampling request is denied; pass `lambda params: True` to restore legacy auto-approve as an explicit opt-in. Requests and denials are logged at WARNING (content is not logged). The per-session counter resets in `_reset_session_state`.
|
||||
- **`MCPTaskOptions`** (experimental, `MCP_LONG_RUNNING_TASKS` feature, **frozen**) - Per-tool-instance options controlling the SEP-2663 long-running task lifecycle. When the server advertises a tool with `execution.taskSupport == "required"`, `MCPTool.call_tool` transparently routes through `call_tool_as_task`, which sends an augmented `tools/call`, polls `tasks/get` until terminal, and reinterprets `tasks/result` as a normal `CallToolResult`. Instances are immutable; replace via `MCPTool.task_options = MCPTaskOptions(...)`. Fields:
|
||||
- `default_ttl: timedelta | None` — forwarded to the server as `params.task.ttl` (milliseconds). When `None`, the server's default applies.
|
||||
|
||||
@@ -9,7 +9,7 @@ integrations, many of which are lazy-loaded from optional packages.
|
||||
"""
|
||||
|
||||
import importlib.metadata
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
from typing import Final
|
||||
|
||||
try:
|
||||
_version = importlib.metadata.version(__name__)
|
||||
@@ -264,7 +264,6 @@ from ._workflows._agent_executor import (
|
||||
)
|
||||
from ._workflows._agent_utils import resolve_agent_id
|
||||
from ._workflows._checkpoint import (
|
||||
CheckpointID,
|
||||
CheckpointStorage,
|
||||
FileCheckpointStorage,
|
||||
InMemoryCheckpointStorage,
|
||||
@@ -308,6 +307,7 @@ from ._workflows._functional import (
|
||||
workflow,
|
||||
)
|
||||
from ._workflows._request_info_mixin import response_handler
|
||||
from ._workflows._runner import Runner
|
||||
from ._workflows._runner_context import (
|
||||
InProcRunnerContext,
|
||||
RunnerContext,
|
||||
@@ -405,7 +405,6 @@ __all__ = [
|
||||
"ChatResponse",
|
||||
"ChatResponseUpdate",
|
||||
"CheckResult",
|
||||
"CheckpointID",
|
||||
"CheckpointStorage",
|
||||
"ClassSkill",
|
||||
"CompactionProvider",
|
||||
@@ -619,20 +618,3 @@ __all__ = [
|
||||
"validate_workflow_graph",
|
||||
"workflow",
|
||||
]
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ._workflows._runner import Runner
|
||||
|
||||
|
||||
def __getattr__(name: str) -> Any:
|
||||
"""Lazily resolve deprecated public names, emitting a ``DeprecationWarning``.
|
||||
|
||||
``Runner`` remains importable from ``agent_framework`` for backward
|
||||
compatibility but is deprecated and slated for removal from the public API.
|
||||
"""
|
||||
if name == "Runner":
|
||||
from ._workflows._runner import Runner, warn_runner_deprecated
|
||||
|
||||
warn_runner_deprecated()
|
||||
return Runner
|
||||
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
||||
|
||||
@@ -74,11 +74,6 @@ _MCP_NORMALIZED_NAME_KEY = "_mcp_normalized_name"
|
||||
# Reserved key in an ``additional_tool_argument_names`` mapping that applies its
|
||||
# values to every tool on the server rather than a single named tool.
|
||||
_MCP_GLOBAL_EXTRA_ARGS_KEY = "*"
|
||||
_MCP_META_LABEL_PATTERN = r"[A-Za-z](?:[A-Za-z0-9-]*[A-Za-z0-9])?"
|
||||
_MCP_META_KEY_PATTERN = re.compile(
|
||||
rf"^(?:(?:{_MCP_META_LABEL_PATTERN})(?:\.{_MCP_META_LABEL_PATTERN})*/)?"
|
||||
r"[A-Za-z0-9](?:[A-Za-z0-9_.-]*[A-Za-z0-9])?$"
|
||||
)
|
||||
# Framework kwargs that flow through the function-invocation pipeline (via
|
||||
# ``FunctionInvocationContext.kwargs``) but must never be forwarded to an MCP
|
||||
# server: they are internal objects that the MCP SDK cannot serialize. They are
|
||||
@@ -210,42 +205,7 @@ def _normalize_additional_tool_argument_names(
|
||||
return set(additional_tool_argument_names), {}
|
||||
|
||||
|
||||
def _mcp_config_candidate_names(*, local_name: str, normalized_name: str, remote_name: str) -> tuple[str, ...]:
|
||||
"""Return safe configuration names for MCP allow/approval matching."""
|
||||
names = [remote_name]
|
||||
if normalized_name == remote_name and local_name != remote_name:
|
||||
names.append(local_name)
|
||||
return tuple(names)
|
||||
|
||||
|
||||
def _validate_mcp_meta_key(key: str) -> None:
|
||||
"""Validate an MCP ``_meta`` key against the 2025-06-18 key-name format."""
|
||||
if not _MCP_META_KEY_PATTERN.fullmatch(key):
|
||||
raise ToolExecutionException(f"Invalid MCP _meta key name: {key!r}.")
|
||||
|
||||
|
||||
def _validate_mcp_meta(raw_meta: object | None) -> dict[str, Any] | None:
|
||||
"""Validate and copy MCP request metadata."""
|
||||
if raw_meta is None:
|
||||
return None
|
||||
if not isinstance(raw_meta, dict):
|
||||
raise ToolExecutionException("MCP tool metadata provided via _meta must be a dict.")
|
||||
|
||||
raw_meta_dict = cast(Mapping[object, Any], raw_meta)
|
||||
meta: dict[str, Any] = {}
|
||||
for key, value in raw_meta_dict.items():
|
||||
if not isinstance(key, str):
|
||||
raise ToolExecutionException("MCP tool metadata provided via _meta must use string keys.")
|
||||
_validate_mcp_meta_key(key)
|
||||
meta[key] = value
|
||||
return meta
|
||||
|
||||
|
||||
def _inject_otel_into_mcp_meta(
|
||||
meta: dict[str, Any] | None = None,
|
||||
*,
|
||||
overwrite: bool = False,
|
||||
) -> dict[str, Any] | None:
|
||||
def _inject_otel_into_mcp_meta(meta: dict[str, Any] | None = None) -> dict[str, Any] | None:
|
||||
"""Inject OpenTelemetry trace context into MCP request _meta via the global propagator(s)."""
|
||||
carrier: dict[str, str] = {}
|
||||
propagate.inject(carrier)
|
||||
@@ -255,8 +215,7 @@ def _inject_otel_into_mcp_meta(
|
||||
if meta is None:
|
||||
meta = {}
|
||||
for key, value in carrier.items():
|
||||
_validate_mcp_meta_key(key)
|
||||
if overwrite or key not in meta:
|
||||
if key not in meta:
|
||||
meta[key] = value
|
||||
|
||||
return meta
|
||||
@@ -422,9 +381,7 @@ class MCPTool:
|
||||
approval_mode: Whether approval is required to run tools.
|
||||
allowed_tools: Optional allow-list of MCP tool names to expose as functions.
|
||||
``None`` (the default) exposes every tool advertised by the MCP server.
|
||||
A non-empty collection exposes only the raw remote tools whose names appear in it. For
|
||||
compatibility, the prefixed local function name is also accepted when the raw remote name already
|
||||
matches its normalized form; normalized aliases do not authorize a different raw remote tool.
|
||||
A non-empty collection exposes only the tools whose names appear in it.
|
||||
An empty collection (``[]``) exposes no tools — if you simply want to
|
||||
disable tool execution, prefer ``load_tools=False`` instead. ``[]`` is
|
||||
useful as a runtime guard or when you want to load tool metadata for
|
||||
@@ -796,14 +753,11 @@ class MCPTool:
|
||||
additional_properties = func.additional_properties or {}
|
||||
normalized_name = additional_properties.get(_MCP_NORMALIZED_NAME_KEY)
|
||||
remote_name = additional_properties.get(_MCP_REMOTE_NAME_KEY)
|
||||
if not isinstance(normalized_name, str) or not isinstance(remote_name, str):
|
||||
continue
|
||||
candidate_names = _mcp_config_candidate_names(
|
||||
local_name=func.name,
|
||||
normalized_name=normalized_name,
|
||||
remote_name=remote_name,
|
||||
)
|
||||
if any(name in allowed_names for name in candidate_names):
|
||||
if (
|
||||
func.name in allowed_names
|
||||
or (isinstance(normalized_name, str) and normalized_name in allowed_names)
|
||||
or (isinstance(remote_name, str) and remote_name in allowed_names)
|
||||
):
|
||||
filtered_functions.append(func)
|
||||
return filtered_functions
|
||||
|
||||
@@ -1427,13 +1381,7 @@ class MCPTool:
|
||||
continue
|
||||
|
||||
input_model = _get_input_model_from_mcp_prompt(prompt)
|
||||
approval_mode = self._determine_approval_mode(
|
||||
*_mcp_config_candidate_names(
|
||||
local_name=local_name,
|
||||
normalized_name=normalized_name,
|
||||
remote_name=prompt.name,
|
||||
)
|
||||
)
|
||||
approval_mode = self._determine_approval_mode(local_name, normalized_name, prompt.name)
|
||||
func: FunctionTool = FunctionTool(
|
||||
func=partial(self.get_prompt, prompt.name),
|
||||
name=local_name,
|
||||
@@ -1474,11 +1422,7 @@ class MCPTool:
|
||||
return
|
||||
|
||||
# Track existing function names to prevent duplicates
|
||||
existing_remote_by_local: dict[str, str] = {}
|
||||
for func in self._functions:
|
||||
remote_name = (func.additional_properties or {}).get(_MCP_REMOTE_NAME_KEY)
|
||||
if isinstance(remote_name, str):
|
||||
existing_remote_by_local[func.name] = remote_name
|
||||
existing_names = {func.name for func in self._functions}
|
||||
tool_call_meta_by_name: dict[str, dict[str, Any]] = {}
|
||||
tool_task_support_by_name: dict[str, str] = {}
|
||||
tool_param_names_by_name: dict[str, set[str]] = {}
|
||||
@@ -1518,7 +1462,7 @@ class MCPTool:
|
||||
|
||||
for tool in tool_list.tools:
|
||||
if tool.meta is not None:
|
||||
tool_call_meta_by_name[tool.name] = _validate_mcp_meta(tool.meta) or {}
|
||||
tool_call_meta_by_name[tool.name] = dict(tool.meta)
|
||||
|
||||
task_support = getattr(getattr(tool, "execution", None), "taskSupport", None)
|
||||
if task_support is not None:
|
||||
@@ -1546,24 +1490,10 @@ class MCPTool:
|
||||
local_name = _build_prefixed_mcp_name(normalized_name, self.tool_name_prefix)
|
||||
|
||||
# Skip if already loaded
|
||||
if local_name in existing_remote_by_local:
|
||||
if existing_remote_by_local.get(local_name) != tool.name:
|
||||
raise ToolExecutionException(
|
||||
"MCP server advertised multiple tools that map to the same local function name: "
|
||||
f"{existing_remote_by_local[local_name]!r} and {tool.name!r} both map to "
|
||||
f"{local_name!r}."
|
||||
)
|
||||
if local_name in existing_names:
|
||||
continue
|
||||
|
||||
existing_remote_by_local[local_name] = tool.name
|
||||
|
||||
approval_mode = self._determine_approval_mode(
|
||||
*_mcp_config_candidate_names(
|
||||
local_name=local_name,
|
||||
normalized_name=normalized_name,
|
||||
remote_name=tool.name,
|
||||
)
|
||||
)
|
||||
approval_mode = self._determine_approval_mode(local_name, normalized_name, tool.name)
|
||||
|
||||
async def _call_tool_with_runtime_kwargs(
|
||||
ctx: FunctionInvocationContext,
|
||||
@@ -1571,13 +1501,8 @@ class MCPTool:
|
||||
_remote_tool_name: str = tool.name,
|
||||
**kwargs: Any,
|
||||
) -> str | list[Content]:
|
||||
trusted_meta = ctx.kwargs.get("_meta")
|
||||
call_kwargs = dict(ctx.kwargs)
|
||||
call_kwargs.update(kwargs)
|
||||
if trusted_meta is not None:
|
||||
call_kwargs["_meta"] = trusted_meta
|
||||
else:
|
||||
call_kwargs.pop("_meta", None)
|
||||
return await self.call_tool(_remote_tool_name, **call_kwargs)
|
||||
|
||||
# Create FunctionTools out of each tool
|
||||
@@ -1593,6 +1518,7 @@ class MCPTool:
|
||||
},
|
||||
)
|
||||
self._functions.append(func)
|
||||
existing_names.add(local_name)
|
||||
|
||||
# Check if there are more pages
|
||||
if not tool_list.nextCursor:
|
||||
@@ -1710,8 +1636,8 @@ class MCPTool:
|
||||
Keyword Args:
|
||||
_meta: Optional ``dict[str, Any]`` of MCP request metadata. This reserved key is passed as the
|
||||
``meta`` parameter of the underlying ``session.call_tool`` call rather than as a tool argument.
|
||||
OpenTelemetry propagation overrides caller-supplied keys, and metadata from ``tools/list``
|
||||
overrides both.
|
||||
User-supplied keys override metadata from ``tools/list``; OpenTelemetry propagation fills in
|
||||
non-conflicting keys.
|
||||
kwargs: Remaining arguments to pass to the tool.
|
||||
|
||||
Returns:
|
||||
@@ -1820,7 +1746,17 @@ class MCPTool:
|
||||
self, tool_name: str, kwargs: dict[str, Any]
|
||||
) -> tuple[dict[str, Any], dict[str, Any] | None]:
|
||||
"""Filter kwargs down to the tool's arguments and build the merged MCP request metadata."""
|
||||
user_meta = _validate_mcp_meta(kwargs.get("_meta"))
|
||||
raw_user_meta: object | None = kwargs.get("_meta")
|
||||
user_meta: dict[str, Any] | None = None
|
||||
if raw_user_meta is not None and not isinstance(raw_user_meta, dict):
|
||||
raise ToolExecutionException("MCP tool metadata provided via _meta must be a dict.")
|
||||
if isinstance(raw_user_meta, dict):
|
||||
raw_user_meta_dict = cast(Mapping[object, object], raw_user_meta)
|
||||
user_meta = {}
|
||||
for key, value in raw_user_meta_dict.items():
|
||||
if not isinstance(key, str):
|
||||
raise ToolExecutionException("MCP tool metadata provided via _meta must use string keys.")
|
||||
user_meta[key] = value
|
||||
|
||||
# Allowlist: forward only the tool's declared parameters (from inputSchema.properties)
|
||||
# plus any user-configured extra argument names. Everything else - notably the
|
||||
@@ -1847,12 +1783,12 @@ class MCPTool:
|
||||
}
|
||||
|
||||
# Some MCP proxies require their tools/list metadata to be echoed on tools/call.
|
||||
request_meta = dict(user_meta) if user_meta is not None else None
|
||||
request_meta = _inject_otel_into_mcp_meta(request_meta, overwrite=True)
|
||||
tool_meta = _validate_mcp_meta(self._tool_call_meta_by_name.get(tool_name))
|
||||
if tool_meta is not None:
|
||||
request_meta = {**(request_meta or {}), **tool_meta}
|
||||
return filtered_kwargs, request_meta
|
||||
tool_meta = self._tool_call_meta_by_name.get(tool_name)
|
||||
request_meta = dict(tool_meta) if tool_meta is not None else None
|
||||
if user_meta is not None:
|
||||
request_meta = {**(request_meta or {}), **user_meta}
|
||||
meta = _inject_otel_into_mcp_meta(request_meta)
|
||||
return filtered_kwargs, meta
|
||||
|
||||
async def call_tool_as_task(self, tool_name: str, **kwargs: Any) -> str | list[Content]:
|
||||
"""Call an MCP tool via the long-running task lifecycle (SEP-2663).
|
||||
|
||||
@@ -2983,8 +2983,6 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
|
||||
self._inner_stream_source: ResponseStream[Any, Any] | Awaitable[ResponseStream[Any, Any]] | None = None
|
||||
self._wrap_inner: bool = False
|
||||
self._map_update: Callable[[Any], UpdateT | Awaitable[UpdateT]] | None = None
|
||||
self._flat_map_update: Callable[[Any], Iterable[UpdateT] | Awaitable[Iterable[UpdateT]]] | None = None
|
||||
self._pending_mapped_updates: list[UpdateT] = []
|
||||
self._pull_context_manager_factories: list[Callable[[], contextlib.AbstractContextManager[Any]]] = []
|
||||
|
||||
def map(
|
||||
@@ -3031,23 +3029,6 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
|
||||
stream._map_update = transform
|
||||
return stream
|
||||
|
||||
def flat_map(
|
||||
self,
|
||||
transform: Callable[[UpdateT], Iterable[OuterUpdateT] | Awaitable[Iterable[OuterUpdateT]]],
|
||||
finalizer: Callable[[Sequence[OuterUpdateT]], OuterFinalT | Awaitable[OuterFinalT]],
|
||||
) -> ResponseStream[OuterUpdateT, OuterFinalT]:
|
||||
"""Create a new stream that transforms each update into zero or more updates.
|
||||
|
||||
Like :meth:`map`, the returned stream delegates iteration to this stream,
|
||||
preserving single consumption and inner finalization/result hooks. Use this
|
||||
when one upstream update naturally expands into multiple wire-protocol events.
|
||||
"""
|
||||
stream: ResponseStream[OuterUpdateT, OuterFinalT] = ResponseStream(self, finalizer=finalizer)
|
||||
stream._inner_stream_source = self
|
||||
stream._wrap_inner = True
|
||||
stream._flat_map_update = transform
|
||||
return stream
|
||||
|
||||
def with_finalizer(
|
||||
self,
|
||||
finalizer: Callable[[Sequence[UpdateT]], OuterFinalT | Awaitable[OuterFinalT]],
|
||||
@@ -3120,7 +3101,35 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
|
||||
def __aiter__(self) -> ResponseStream[UpdateT, FinalT]:
|
||||
return self
|
||||
|
||||
async def _record_update(self, update: UpdateT) -> UpdateT:
|
||||
async def __anext__(self) -> UpdateT:
|
||||
try:
|
||||
with contextlib.ExitStack() as stack:
|
||||
for factory in self._pull_context_manager_factories:
|
||||
stack.enter_context(factory())
|
||||
# Resolve the underlying stream inside the pull contexts so that any
|
||||
# spans/contexts created during stream resolution (e.g. inner chat
|
||||
# completion spans created on the first pull of a wrapped agent stream)
|
||||
# inherit the active context (e.g. an outer agent invoke span).
|
||||
if self._iterator is None:
|
||||
stream = await self._get_stream()
|
||||
self._iterator = stream.__aiter__()
|
||||
update: UpdateT = await self._iterator.__anext__()
|
||||
except StopAsyncIteration:
|
||||
self._consumed = True
|
||||
await self._run_cleanup_hooks()
|
||||
await self.get_final_response()
|
||||
raise
|
||||
except Exception as exc:
|
||||
self._stream_error = exc
|
||||
try:
|
||||
await self._run_cleanup_hooks()
|
||||
finally:
|
||||
self._stream_error = None
|
||||
raise
|
||||
if self._map_update is not None:
|
||||
update = self._map_update(update) # type: ignore[assignment]
|
||||
if isawaitable(update):
|
||||
update = await update
|
||||
self._updates.append(update)
|
||||
for hook in self._transform_hooks:
|
||||
hooked = hook(update)
|
||||
@@ -3130,47 +3139,6 @@ class ResponseStream(AsyncIterable[UpdateT], Generic[UpdateT, FinalT]):
|
||||
update = cast(UpdateT, hooked)
|
||||
return update
|
||||
|
||||
async def __anext__(self) -> UpdateT:
|
||||
while True:
|
||||
if self._pending_mapped_updates:
|
||||
return await self._record_update(self._pending_mapped_updates.pop(0))
|
||||
|
||||
try:
|
||||
with contextlib.ExitStack() as stack:
|
||||
for factory in self._pull_context_manager_factories:
|
||||
stack.enter_context(factory())
|
||||
# Resolve the underlying stream inside the pull contexts so that any
|
||||
# spans/contexts created during stream resolution (e.g. inner chat
|
||||
# completion spans created on the first pull of a wrapped agent stream)
|
||||
# inherit the active context (e.g. an outer agent invoke span).
|
||||
if self._iterator is None:
|
||||
stream = await self._get_stream()
|
||||
self._iterator = stream.__aiter__()
|
||||
update: UpdateT = await self._iterator.__anext__()
|
||||
except StopAsyncIteration:
|
||||
self._consumed = True
|
||||
await self._run_cleanup_hooks()
|
||||
await self.get_final_response()
|
||||
raise
|
||||
except Exception as exc:
|
||||
self._stream_error = exc
|
||||
try:
|
||||
await self._run_cleanup_hooks()
|
||||
finally:
|
||||
self._stream_error = None
|
||||
raise
|
||||
if self._flat_map_update is not None:
|
||||
mapped_updates = self._flat_map_update(update)
|
||||
if isawaitable(mapped_updates):
|
||||
mapped_updates = await mapped_updates
|
||||
self._pending_mapped_updates.extend(mapped_updates)
|
||||
continue
|
||||
if self._map_update is not None:
|
||||
update = self._map_update(update) # type: ignore[assignment]
|
||||
if isawaitable(update):
|
||||
update = await update
|
||||
return await self._record_update(update)
|
||||
|
||||
async def _resolve_stream_with_pull_contexts(self) -> AsyncIterable[UpdateT]:
|
||||
"""Resolve the underlying stream while activating any registered pull context managers.
|
||||
|
||||
|
||||
@@ -3,7 +3,6 @@
|
||||
import asyncio
|
||||
import contextlib
|
||||
import logging
|
||||
import warnings
|
||||
from collections import defaultdict
|
||||
from collections.abc import AsyncGenerator, Sequence
|
||||
from typing import Any
|
||||
@@ -11,6 +10,7 @@ from typing import Any
|
||||
from ..exceptions import (
|
||||
WorkflowCheckpointException,
|
||||
WorkflowConvergenceException,
|
||||
WorkflowRunnerException,
|
||||
)
|
||||
from ._checkpoint import CheckpointID, CheckpointStorage, WorkflowCheckpoint
|
||||
from ._const import EXECUTOR_STATE_KEY
|
||||
@@ -27,21 +27,6 @@ from ._state import State
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def warn_runner_deprecated() -> None:
|
||||
"""Emit a deprecation warning when ``Runner`` is accessed from the public API.
|
||||
|
||||
``Runner`` remains importable from ``agent_framework`` for backward
|
||||
compatibility, but it is intended for internal use only and will be removed
|
||||
from the public API in a future version.
|
||||
"""
|
||||
warnings.warn(
|
||||
"`Runner` is deprecated and will be removed from the public API in a future version. "
|
||||
"It is intended for internal use only.",
|
||||
DeprecationWarning,
|
||||
stacklevel=3,
|
||||
)
|
||||
|
||||
|
||||
class Runner:
|
||||
"""A class to run a workflow in Pregel supersteps."""
|
||||
|
||||
@@ -78,34 +63,25 @@ class Runner:
|
||||
self._iteration = 0
|
||||
self._max_iterations = max_iterations
|
||||
self._state = state
|
||||
|
||||
# Checkpointing related attributes
|
||||
self._resumed_from_checkpoint = False
|
||||
self._previous_checkpoint_id: CheckpointID | None = None
|
||||
self._running = False
|
||||
self._resumed_from_checkpoint = False # Track whether we resumed
|
||||
|
||||
@property
|
||||
def context(self) -> RunnerContext:
|
||||
"""Get the runner context for message, event, and checkpoint handling."""
|
||||
"""Get the workflow context."""
|
||||
return self._ctx
|
||||
|
||||
@property
|
||||
def state(self) -> State:
|
||||
"""Get the shared state for the workflow."""
|
||||
return self._state
|
||||
|
||||
def reset_iteration_count(self) -> None:
|
||||
"""Reset the iteration count to zero.
|
||||
|
||||
This is useful when the workflow resumes from a new set of messages.
|
||||
|
||||
Note:
|
||||
When a workflow is resumed from a response (for a request_info_event)
|
||||
or a checkpoint, the iteration count is normally NOT reset.
|
||||
"""
|
||||
"""Reset the iteration count to zero."""
|
||||
self._iteration = 0
|
||||
|
||||
async def run_until_convergence(self) -> AsyncGenerator[WorkflowEvent, None]:
|
||||
"""Run the workflow until no more messages are sent."""
|
||||
if self._running:
|
||||
raise WorkflowRunnerException("Runner is already running.")
|
||||
|
||||
self._running = True
|
||||
previous_checkpoint_id: CheckpointID | None = None
|
||||
try:
|
||||
# Emit any events already produced prior to entering loop
|
||||
if await self._ctx.has_events():
|
||||
@@ -113,12 +89,12 @@ class Runner:
|
||||
for event in await self._ctx.drain_events():
|
||||
yield event
|
||||
|
||||
# Create a checkpoint before a run starts. Checkpoints are usually considered to be created at the
|
||||
# end of an iteration, we can think of this checkpoint as being created at the end of "superstep 0"
|
||||
# which captures the states after which the start executor has run. Note that we execute the start
|
||||
# executor outside of the main iteration loop.
|
||||
if await self._ctx.has_messages() and self._iteration == 0 and not self._resumed_from_checkpoint:
|
||||
await self.create_checkpoint_if_enabled()
|
||||
# Create the first checkpoint. Checkpoints are usually considered to be created at the end of an iteration,
|
||||
# we can think of the first checkpoint as being created at the end of a "superstep 0" which captures the
|
||||
# states after which the start executor has run. Note that we execute the start executor outside of the
|
||||
# main iteration loop.
|
||||
if await self._ctx.has_messages() and not self._resumed_from_checkpoint:
|
||||
previous_checkpoint_id = await self._create_checkpoint_if_enabled(previous_checkpoint_id)
|
||||
|
||||
while self._iteration < self._max_iterations:
|
||||
logger.info(f"Starting superstep {self._iteration + 1}")
|
||||
@@ -165,7 +141,7 @@ class Runner:
|
||||
self._state.commit()
|
||||
|
||||
# Create checkpoint after each superstep iteration
|
||||
await self.create_checkpoint_if_enabled()
|
||||
previous_checkpoint_id = await self._create_checkpoint_if_enabled(previous_checkpoint_id)
|
||||
|
||||
yield WorkflowEvent.superstep_completed(iteration=self._iteration)
|
||||
|
||||
@@ -173,15 +149,13 @@ class Runner:
|
||||
if not await self._ctx.has_messages():
|
||||
break
|
||||
|
||||
logger.info(f"Workflow completed after {self._iteration} supersteps")
|
||||
|
||||
if self._iteration >= self._max_iterations and await self._ctx.has_messages():
|
||||
raise WorkflowConvergenceException(f"Runner did not converge after {self._max_iterations} iterations.")
|
||||
|
||||
logger.info(f"Workflow completed after {self._iteration} supersteps")
|
||||
self._resumed_from_checkpoint = False # Reset resume flag for next run
|
||||
finally:
|
||||
# Reset the resume flag so stale resume state never leaks into the next run on this
|
||||
# instance - even if convergence raised before completing (e.g. an executor failure
|
||||
# during a resumed run).
|
||||
self._resumed_from_checkpoint = False
|
||||
self._running = False
|
||||
|
||||
async def _run_iteration(self) -> None:
|
||||
"""Run a single iteration of the workflow.
|
||||
@@ -235,55 +209,40 @@ class Runner:
|
||||
]
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
async def _prepare_checkpoint_state(self) -> None:
|
||||
"""Persist executor snapshots into committed shared state.
|
||||
|
||||
This is used by checkpoint capture paths that need a complete, restorable
|
||||
state payload without necessarily writing to a checkpoint storage backend.
|
||||
"""
|
||||
await self._save_executor_states()
|
||||
self._state.commit()
|
||||
|
||||
async def create_checkpoint_if_enabled(self) -> None:
|
||||
async def _create_checkpoint_if_enabled(self, previous_checkpoint_id: CheckpointID | None) -> CheckpointID | None:
|
||||
"""Create a checkpoint if checkpointing is enabled and attach a label and metadata."""
|
||||
if not self._ctx.has_checkpointing():
|
||||
return
|
||||
return None
|
||||
|
||||
try:
|
||||
# Save executor states into committed state before creating the checkpoint.
|
||||
await self._prepare_checkpoint_state()
|
||||
# Save executor states into the shared state before creating the checkpoint,
|
||||
# so that they are included in the checkpoint payload.
|
||||
await self._save_executor_states()
|
||||
# `on_checkpoint_save()` writes via State.set(), which stages values in the
|
||||
# pending buffer. Checkpoints serialize committed state only, so commit here
|
||||
# to ensure executor snapshots are captured in this checkpoint.
|
||||
self._state.commit()
|
||||
|
||||
checkpoint_id = await self._ctx.create_checkpoint(
|
||||
self._workflow_name,
|
||||
self._graph_signature_hash,
|
||||
self._state,
|
||||
self._previous_checkpoint_id,
|
||||
previous_checkpoint_id,
|
||||
self._iteration,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Created checkpoint: %s with parent checkpoint at iteration %d: %s",
|
||||
checkpoint_id,
|
||||
self._iteration,
|
||||
self._previous_checkpoint_id,
|
||||
)
|
||||
self._previous_checkpoint_id = checkpoint_id
|
||||
logger.info(f"Created checkpoint: {checkpoint_id}")
|
||||
return checkpoint_id
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Failed to create checkpoint at iteration %d: %s. "
|
||||
"Note that this does not fail the workflow run. "
|
||||
"The next successfully-created checkpoint will be parented to the last successful checkpoint: %s",
|
||||
self._iteration,
|
||||
e,
|
||||
self._previous_checkpoint_id,
|
||||
)
|
||||
logger.warning(f"Failed to create checkpoint: {e}")
|
||||
return None
|
||||
|
||||
async def restore_from_checkpoint(
|
||||
self,
|
||||
checkpoint_id: CheckpointID,
|
||||
checkpoint_storage: CheckpointStorage | None = None,
|
||||
) -> None:
|
||||
"""Restore the runner from a checkpoint.
|
||||
"""Restore workflow state from a checkpoint.
|
||||
|
||||
Args:
|
||||
checkpoint_id: The ID of the checkpoint to restore from
|
||||
@@ -331,7 +290,7 @@ class Runner:
|
||||
# Apply the checkpoint to the context
|
||||
await self._ctx.apply_checkpoint(checkpoint)
|
||||
# Mark the runner as resumed
|
||||
self._mark_resumed(checkpoint)
|
||||
self._mark_resumed(checkpoint.iteration_count)
|
||||
|
||||
logger.info(f"Successfully restored workflow from checkpoint: {checkpoint_id}")
|
||||
except WorkflowCheckpointException:
|
||||
@@ -397,14 +356,13 @@ class Runner:
|
||||
|
||||
return parsed
|
||||
|
||||
def _mark_resumed(self, checkpoint: WorkflowCheckpoint) -> None:
|
||||
def _mark_resumed(self, iteration: int) -> None:
|
||||
"""Mark the runner as having resumed from a checkpoint.
|
||||
|
||||
Optionally set the current iteration and max iterations.
|
||||
"""
|
||||
self._resumed_from_checkpoint = True
|
||||
self._iteration = checkpoint.iteration_count
|
||||
self._previous_checkpoint_id = checkpoint.checkpoint_id
|
||||
self._iteration = iteration
|
||||
|
||||
async def _set_executor_state(self, executor_id: str, state: dict[str, Any]) -> None:
|
||||
"""Store executor state in state under a reserved key.
|
||||
|
||||
@@ -11,14 +11,12 @@ import logging
|
||||
import types
|
||||
import uuid
|
||||
import warnings
|
||||
import weakref
|
||||
from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence
|
||||
from dataclasses import dataclass, field
|
||||
from typing import TYPE_CHECKING, Any, Literal, overload
|
||||
|
||||
from .._sessions import ContextProvider
|
||||
from .._types import ResponseStream
|
||||
from ..exceptions import WorkflowException
|
||||
from ..observability import OtelAttr, capture_exception, create_workflow_span
|
||||
from ._checkpoint import CheckpointStorage
|
||||
from ._const import DEFAULT_MAX_ITERATIONS, GLOBAL_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY
|
||||
@@ -348,29 +346,25 @@ class Workflow(DictConvertible):
|
||||
# Store non-serializable runtime objects as private attributes
|
||||
self._runner_context = runner_context
|
||||
self._runner_context.set_yield_output_classifier(self._output_designation.classify)
|
||||
self._state = State()
|
||||
self._runner: Runner = Runner(
|
||||
self.edge_groups,
|
||||
self.executors,
|
||||
State(),
|
||||
self._state,
|
||||
runner_context,
|
||||
self.name,
|
||||
self.graph_signature_hash,
|
||||
max_iterations=max_iterations,
|
||||
)
|
||||
|
||||
# Flag to prevent concurrent workflow executions
|
||||
self._is_running = False
|
||||
|
||||
# Current run-level status of this workflow instance. Updated in lockstep with
|
||||
# the status events emitted from `_run_workflow_with_tracing`. Defaults to IDLE
|
||||
# for a freshly built workflow that has not yet been run.
|
||||
self._status: WorkflowRunState = WorkflowRunState.IDLE
|
||||
|
||||
# Weak reference to the in-flight run's ``ResponseStream``. Used as the single
|
||||
# concurrency lock: if the previous stream is still alive, ``run()`` rejects a
|
||||
# new run synchronously (before any await). When the stream is fully consumed
|
||||
# ``_run_core``'s finally clears this; if the caller drops the stream without
|
||||
# ever iterating, the weakref dereferences to ``None`` once Python collects it,
|
||||
# so a subsequent ``run()`` is allowed.
|
||||
self._active_run: weakref.ref[ResponseStream[WorkflowEvent, WorkflowRunResult]] | None = None
|
||||
|
||||
@property
|
||||
def status(self) -> WorkflowRunState:
|
||||
"""Return the current run-level status of this workflow instance.
|
||||
@@ -382,6 +376,16 @@ class Workflow(DictConvertible):
|
||||
"""
|
||||
return self._status
|
||||
|
||||
def _ensure_not_running(self) -> None:
|
||||
"""Ensure the workflow is not already running."""
|
||||
if self._is_running:
|
||||
raise RuntimeError("Workflow is already running. Concurrent executions are not allowed.")
|
||||
self._is_running = True
|
||||
|
||||
def _reset_running_flag(self) -> None:
|
||||
"""Reset the running flag."""
|
||||
self._is_running = False
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Serialize the workflow definition into a JSON-ready dictionary."""
|
||||
data: dict[str, Any] = {
|
||||
@@ -531,12 +535,13 @@ class Workflow(DictConvertible):
|
||||
yield in_progress # noqa: RUF070
|
||||
|
||||
# Per-run reset for fresh-message runs only. We deliberately
|
||||
# do NOT clear shared workflow state or the runner context's
|
||||
# in-flight messages here - state and pending work persist
|
||||
# across `run()` calls so that a `WorkflowAgent` can deliver
|
||||
# multi-turn input on the same instance and have prior turns'
|
||||
# context survive. Iteration counting and per-run kwargs ARE
|
||||
# per-run though, so they're reset here.
|
||||
# do NOT clear shared workflow state (`_state.clear()`) or the
|
||||
# runner context's in-flight messages (`reset_for_new_run()`)
|
||||
# here - state and pending work persist across `run()` calls
|
||||
# so that a `WorkflowAgent` can deliver multi-turn input on
|
||||
# the same instance and have prior turns' context survive.
|
||||
# Iteration counting and per-run kwargs ARE per-run though,
|
||||
# so they're reset here.
|
||||
if not is_continuation:
|
||||
self._runner.reset_iteration_count()
|
||||
|
||||
@@ -559,13 +564,14 @@ class Workflow(DictConvertible):
|
||||
combined_kwargs["client_kwargs"] = self._resolve_invocation_kwargs(
|
||||
client_kwargs, "client_kwargs"
|
||||
)
|
||||
self._runner.state.set(WORKFLOW_RUN_KWARGS_KEY, combined_kwargs)
|
||||
self._state.set(WORKFLOW_RUN_KWARGS_KEY, combined_kwargs)
|
||||
elif not is_continuation:
|
||||
self._runner.state.set(WORKFLOW_RUN_KWARGS_KEY, {})
|
||||
self._runner.state.commit() # Commit immediately so kwargs are available
|
||||
self._state.set(WORKFLOW_RUN_KWARGS_KEY, {})
|
||||
self._state.commit() # Commit immediately so kwargs are available
|
||||
|
||||
# Explicitly set streaming mode per run
|
||||
self._runner.context.set_streaming(streaming)
|
||||
# Set streaming mode (always set explicitly per run since
|
||||
# reset_for_new_run() no longer runs to clear it).
|
||||
self._runner_context.set_streaming(streaming)
|
||||
|
||||
# Execute initial setup if provided
|
||||
if initial_executor_fn:
|
||||
@@ -659,7 +665,7 @@ class Workflow(DictConvertible):
|
||||
await executor.execute(
|
||||
message,
|
||||
[self.__class__.__name__],
|
||||
self._runner.state,
|
||||
self._state,
|
||||
self._runner.context,
|
||||
trace_contexts=None,
|
||||
source_span_ids=None,
|
||||
@@ -739,28 +745,9 @@ class Workflow(DictConvertible):
|
||||
Raises:
|
||||
ValueError: If parameter combination is invalid.
|
||||
"""
|
||||
# Validate parameters first so misuse fails before we touch any run state.
|
||||
# Validate parameters and set running flag eagerly (before any async work)
|
||||
self._validate_run_params(message, responses, checkpoint_id)
|
||||
|
||||
# Concurrency check: reject a second run synchronously - before constructing
|
||||
# the ResponseStream or yielding control to the event loop - so a concurrent
|
||||
# ``run`` call can't slip past the guard while the first call is suspended
|
||||
# inside its async generator. The ``ResponseStream`` returned below is the
|
||||
# lock: as long as the caller holds a reference to it, ``self._active_run()``
|
||||
# resolves to a live object and a new ``run`` is rejected. When the stream is
|
||||
# fully consumed, ``_run_core``'s finally clears the attribute. When the
|
||||
# caller drops the stream without iterating, garbage collection invalidates
|
||||
# the weakref, so a subsequent ``run`` is permitted.
|
||||
if self._is_run_active():
|
||||
raise WorkflowException(
|
||||
"Workflow is already running; concurrent runs are not allowed on the same instance."
|
||||
)
|
||||
|
||||
# No run is active, so any runtime checkpoint storage override still set on the
|
||||
# context is stale - left over from a prior run whose stream was dropped before
|
||||
# its async-generator finalizer ran. Clear it so this run starts clean and does
|
||||
# not silently inherit the prior run's runtime checkpoint storage.
|
||||
self._runner.context.clear_runtime_checkpoint_storage()
|
||||
self._ensure_not_running()
|
||||
|
||||
response_stream = ResponseStream[WorkflowEvent, WorkflowRunResult](
|
||||
self._run_core(
|
||||
@@ -773,8 +760,10 @@ class Workflow(DictConvertible):
|
||||
client_kwargs=client_kwargs,
|
||||
),
|
||||
finalizer=functools.partial(self._finalize_events, include_status_events=include_status_events),
|
||||
cleanup_hooks=[
|
||||
functools.partial(self._run_cleanup, checkpoint_storage),
|
||||
],
|
||||
)
|
||||
self._active_run = weakref.ref(response_stream)
|
||||
|
||||
if stream:
|
||||
return response_stream
|
||||
@@ -796,79 +785,55 @@ class Workflow(DictConvertible):
|
||||
Yields:
|
||||
WorkflowEvent: The events generated during the workflow execution.
|
||||
"""
|
||||
# Capture the weakref instance ``run()`` installed for *this* run. We
|
||||
# compare by object identity in the finally so a stale finalizer (e.g.
|
||||
# the caller dropped this stream after partial iteration, then started
|
||||
# a new run before async-gen finalization throws ``GeneratorExit`` into
|
||||
# us) does not clobber a successor run's freshly installed weakref.
|
||||
# ``run()`` runs synchronously and assigns ``self._active_run`` before
|
||||
# this generator's body is first iterated, so by the time we read it
|
||||
# here it already points at our own ``ResponseStream``.
|
||||
my_active_run = self._active_run
|
||||
|
||||
# Enable runtime checkpointing if storage provided.
|
||||
# Enable runtime checkpointing if storage provided
|
||||
if checkpoint_storage is not None:
|
||||
self._runner.context.set_runtime_checkpoint_storage(checkpoint_storage)
|
||||
|
||||
try:
|
||||
# Async validation: a fresh-message run is only allowed when the
|
||||
# runner context has fully drained from any prior run. If it still
|
||||
# has in-flight executor messages, the prior run didn't complete -
|
||||
# the caller must either resume from a checkpoint or wait for the
|
||||
# prior run to drain. (Pending request_info events are intentionally
|
||||
# NOT blocked here: a follow-up run with message=... is the normal
|
||||
# way to deliver a response to those pending requests, e.g. via
|
||||
# WorkflowAgent._process_pending_requests.)
|
||||
# NOTE: _validate_run_params already enforces that ``message`` is
|
||||
# mutually exclusive with both ``checkpoint_id`` and ``responses``,
|
||||
# so we don't need to re-check those here.
|
||||
if message is not None and await self._runner.context.has_messages():
|
||||
raise RuntimeError(
|
||||
"Cannot start a new run with 'message' while in-flight executor "
|
||||
"messages remain from a prior run. Resume from a checkpoint "
|
||||
"(checkpoint_id=...) or wait for the prior run to complete. "
|
||||
"Workflows that need to recover from a mid-run failure must use "
|
||||
"checkpointing; there is no in-process recovery path."
|
||||
)
|
||||
# Async validation: a fresh-message run is only allowed when the
|
||||
# runner context has fully drained from any prior run. If it still
|
||||
# has in-flight executor messages, the prior run didn't complete -
|
||||
# the caller must either resume from a checkpoint or wait for the
|
||||
# prior run to drain. (Pending request_info events are intentionally
|
||||
# NOT blocked here: a follow-up run with message=... is the normal
|
||||
# way to deliver a response to those pending requests, e.g. via
|
||||
# WorkflowAgent._process_pending_requests.)
|
||||
# NOTE: _validate_run_params already enforces that ``message`` is
|
||||
# mutually exclusive with both ``checkpoint_id`` and ``responses``,
|
||||
# so we don't need to re-check those here.
|
||||
if message is not None and await self._runner.context.has_messages():
|
||||
raise RuntimeError(
|
||||
"Cannot start a new run with 'message' while in-flight executor "
|
||||
"messages remain from a prior run. Resume from a checkpoint "
|
||||
"(checkpoint_id=...) or wait for the prior run to complete. "
|
||||
"Workflows that need to recover from a mid-run failure must use "
|
||||
"checkpointing; there is no in-process recovery path."
|
||||
)
|
||||
|
||||
initial_executor_fn = self._resolve_execution_mode(message, responses, checkpoint_id, checkpoint_storage)
|
||||
initial_executor_fn = self._resolve_execution_mode(message, responses, checkpoint_id, checkpoint_storage)
|
||||
|
||||
async for event in self._run_workflow_with_tracing(
|
||||
initial_executor_fn=initial_executor_fn,
|
||||
is_continuation=(message is None),
|
||||
streaming=streaming,
|
||||
function_invocation_kwargs=function_invocation_kwargs,
|
||||
client_kwargs=client_kwargs,
|
||||
):
|
||||
if event.type == "request_info" and event.request_id in (responses or {}):
|
||||
# Don't yield request_info events for which we have responses to send -
|
||||
# these are considered "handled". This prevents the caller from seeing
|
||||
# events for requests they are already responding to.
|
||||
# This usually happens when responses are provided with a checkpoint
|
||||
# (restore then send), because the request_info events are stored in the
|
||||
# checkpoint and would be emitted on restoration by the runner regardless
|
||||
# of if a response is provided or not.
|
||||
continue
|
||||
yield event
|
||||
finally:
|
||||
# Whether this run is still the active one (no successor ``run()`` has
|
||||
# installed a new weakref since we started). Captured once because the
|
||||
# active-run clear below mutates ``self._active_run``. Used to scope both
|
||||
# the run-lock release and the runtime-storage clear so a dropped run's
|
||||
# deferred finalizer cannot clobber a successor run's state.
|
||||
owns_run = self._active_run is my_active_run
|
||||
if owns_run:
|
||||
# Clear the active-run weakref so a subsequent ``run()`` is allowed.
|
||||
# If the caller dropped this stream after partial iteration and a new
|
||||
# ``run()`` already installed its own weakref before our async-gen
|
||||
# finalizer ran, ``self._active_run`` points at the successor and we
|
||||
# leave it untouched to preserve the successor's concurrency guard.
|
||||
self._active_run = None
|
||||
# Same ownership scoping applies to the runtime checkpoint storage:
|
||||
# only clear it when this run still owns it, so a dropped run's
|
||||
# deferred finalizer can't clear a successor's storage.
|
||||
if checkpoint_storage is not None:
|
||||
self._runner.context.clear_runtime_checkpoint_storage()
|
||||
async for event in self._run_workflow_with_tracing(
|
||||
initial_executor_fn=initial_executor_fn,
|
||||
is_continuation=(message is None),
|
||||
streaming=streaming,
|
||||
function_invocation_kwargs=function_invocation_kwargs,
|
||||
client_kwargs=client_kwargs,
|
||||
):
|
||||
if event.type == "request_info" and event.request_id in (responses or {}):
|
||||
# Don't yield request_info events for which we have responses to send -
|
||||
# these are considered "handled". This prevents the caller from seeing
|
||||
# events for requests they are already responding to.
|
||||
# This usually happens when responses are provided with a checkpoint
|
||||
# (restore then send), because the request_info events are stored in the
|
||||
# checkpoint and would be emitted on restoration by the runner regardless
|
||||
# of if a response is provided or not.
|
||||
continue
|
||||
yield event
|
||||
|
||||
async def _run_cleanup(self, checkpoint_storage: CheckpointStorage | None) -> None:
|
||||
"""Cleanup hook called after stream consumption."""
|
||||
if checkpoint_storage is not None:
|
||||
self._runner.context.clear_runtime_checkpoint_storage()
|
||||
self._reset_running_flag()
|
||||
|
||||
@staticmethod
|
||||
def _finalize_events(
|
||||
@@ -970,7 +935,7 @@ class Workflow(DictConvertible):
|
||||
|
||||
async def _send_responses_internal(self, responses: Mapping[str, Any]) -> None:
|
||||
"""Internal method to validate and send responses to the executors."""
|
||||
pending_requests = await self._runner.context.get_pending_request_info_events()
|
||||
pending_requests = await self._runner_context.get_pending_request_info_events()
|
||||
if not pending_requests:
|
||||
raise RuntimeError("No pending requests found in workflow context.")
|
||||
|
||||
@@ -990,7 +955,7 @@ class Workflow(DictConvertible):
|
||||
coerced_responses[request_id] = response
|
||||
|
||||
await asyncio.gather(*[
|
||||
self._runner.context.send_request_info_response(request_id, response)
|
||||
self._runner_context.send_request_info_response(request_id, response)
|
||||
for request_id, response in coerced_responses.items()
|
||||
])
|
||||
|
||||
@@ -1186,12 +1151,3 @@ class Workflow(DictConvertible):
|
||||
context_providers=context_providers,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
def _is_run_active(self) -> bool:
|
||||
"""Check if a workflow run is currently active.
|
||||
|
||||
Returns:
|
||||
True if a run is active, False otherwise.
|
||||
"""
|
||||
existing_stream = self._active_run() if self._active_run is not None else None
|
||||
return existing_stream is not None
|
||||
|
||||
@@ -122,113 +122,6 @@ async def test_load_tools_with_tool_name_prefix_preserves_matching_configuration
|
||||
assert tool.functions[0].approval_mode == "always_require"
|
||||
|
||||
|
||||
async def test_allowed_tools_does_not_authorize_normalized_remote_name_collision() -> None:
|
||||
"""A normalized/local allowlist match must not authorize a different raw remote tool."""
|
||||
tool = MCPTool(name="test_server", allowed_tools=["delete-file"]) # type: ignore[abstract]
|
||||
|
||||
mock_session = AsyncMock()
|
||||
tool.session = mock_session
|
||||
tool.load_tools_flag = True
|
||||
|
||||
page = Mock()
|
||||
page.tools = [
|
||||
types.Tool(
|
||||
name="delete/file",
|
||||
description="Delete a file",
|
||||
inputSchema={"type": "object", "properties": {}},
|
||||
),
|
||||
]
|
||||
page.nextCursor = None
|
||||
mock_session.list_tools = AsyncMock(return_value=page)
|
||||
|
||||
await tool.load_tools()
|
||||
|
||||
assert [function.name for function in tool._functions] == ["delete-file"]
|
||||
assert tool.functions == []
|
||||
|
||||
|
||||
async def test_load_tools_rejects_colliding_normalized_tool_names() -> None:
|
||||
"""A remote MCP server must not choose which raw tool backs a colliding local name."""
|
||||
tool = MCPTool(name="test_server", allowed_tools=["delete-file"]) # type: ignore[abstract]
|
||||
|
||||
mock_session = AsyncMock()
|
||||
tool.session = mock_session
|
||||
tool.load_tools_flag = True
|
||||
|
||||
page = Mock()
|
||||
page.tools = [
|
||||
types.Tool(
|
||||
name="delete/file",
|
||||
description="Unauthorized tool",
|
||||
inputSchema={"type": "object", "properties": {}},
|
||||
),
|
||||
types.Tool(
|
||||
name="delete-file",
|
||||
description="Authorized tool",
|
||||
inputSchema={"type": "object", "properties": {}},
|
||||
),
|
||||
]
|
||||
page.nextCursor = None
|
||||
mock_session.list_tools = AsyncMock(return_value=page)
|
||||
|
||||
with pytest.raises(ToolExecutionException, match="map to the same local function name"):
|
||||
await tool.load_tools()
|
||||
|
||||
|
||||
async def test_allowed_tools_exact_raw_name_allows_normalized_function_name() -> None:
|
||||
"""An exact raw remote allowlist entry still exposes that raw tool, regardless of local normalization."""
|
||||
tool = MCPTool(name="test_server", allowed_tools=["delete/file"]) # type: ignore[abstract]
|
||||
|
||||
mock_session = AsyncMock()
|
||||
tool.session = mock_session
|
||||
tool.load_tools_flag = True
|
||||
|
||||
page = Mock()
|
||||
page.tools = [
|
||||
types.Tool(
|
||||
name="delete/file",
|
||||
description="Delete a file",
|
||||
inputSchema={"type": "object", "properties": {}},
|
||||
),
|
||||
]
|
||||
page.nextCursor = None
|
||||
mock_session.list_tools = AsyncMock(return_value=page)
|
||||
|
||||
await tool.load_tools()
|
||||
|
||||
assert [function.name for function in tool.functions] == ["delete-file"]
|
||||
assert tool.functions[0].additional_properties is not None
|
||||
assert tool.functions[0].additional_properties["_mcp_remote_name"] == "delete/file"
|
||||
|
||||
|
||||
async def test_approval_mode_does_not_match_normalized_colliding_name() -> None:
|
||||
"""Approval rules should not apply to a different raw remote tool through normalization."""
|
||||
tool = MCPTool( # type: ignore[abstract]
|
||||
name="test_server",
|
||||
approval_mode={"always_require_approval": ["delete-file"]},
|
||||
)
|
||||
|
||||
mock_session = AsyncMock()
|
||||
tool.session = mock_session
|
||||
tool.load_tools_flag = True
|
||||
|
||||
page = Mock()
|
||||
page.tools = [
|
||||
types.Tool(
|
||||
name="delete/file",
|
||||
description="Delete a file",
|
||||
inputSchema={"type": "object", "properties": {}},
|
||||
),
|
||||
]
|
||||
page.nextCursor = None
|
||||
mock_session.list_tools = AsyncMock(return_value=page)
|
||||
|
||||
await tool.load_tools()
|
||||
|
||||
assert tool._functions[0].name == "delete-file"
|
||||
assert tool._functions[0].approval_mode == "never_require"
|
||||
|
||||
|
||||
async def test_load_prompts_with_tool_name_prefix() -> None:
|
||||
"""Prefixed MCP prompt names should be exposed with the configured prefix."""
|
||||
tool = MCPTool(name="docs", tool_name_prefix="docs") # type: ignore[abstract]
|
||||
@@ -3446,7 +3339,6 @@ async def test_load_tools_adds_properties_to_zero_arg_tool_schema():
|
||||
none_schema_tool.name = "none_schema_tool"
|
||||
none_schema_tool.description = "A tool with None inputSchema"
|
||||
none_schema_tool.inputSchema = None
|
||||
none_schema_tool.meta = None
|
||||
page.tools.append(none_schema_tool)
|
||||
page.nextCursor = None
|
||||
|
||||
@@ -4885,7 +4777,7 @@ async def test_mcp_tool_call_tool_forwards_tool_list_meta():
|
||||
|
||||
|
||||
async def test_mcp_tool_call_tool_user_meta_merges_with_tool_list_meta():
|
||||
"""Tools/list _meta should win over caller-provided _meta on conflicts."""
|
||||
"""User-provided _meta should be sent as MCP request metadata, not tool arguments."""
|
||||
from opentelemetry import trace
|
||||
|
||||
tool_meta = {"from_tool": "tool-value", "shared": "tool-value"}
|
||||
@@ -4925,153 +4817,11 @@ async def test_mcp_tool_call_tool_user_meta_merges_with_tool_list_meta():
|
||||
assert call_kwargs["meta"] == {
|
||||
"from_tool": "tool-value",
|
||||
"from_user": "user-value",
|
||||
"shared": "tool-value",
|
||||
"shared": "user-value",
|
||||
}
|
||||
assert user_meta == {"from_user": "user-value", "shared": "user-value"}
|
||||
|
||||
|
||||
async def test_mcp_tool_function_invocation_strips_model_supplied_meta() -> None:
|
||||
"""Model-supplied _meta should not become MCP request metadata."""
|
||||
from opentelemetry import trace
|
||||
|
||||
class TestServer(MCPTool):
|
||||
async def connect(self) -> None: # type: ignore[override] # pyrefly: ignore[bad-override] # ty: ignore[invalid-method-override]
|
||||
self.session = Mock(spec=ClientSession)
|
||||
self.session.list_tools = AsyncMock(
|
||||
return_value=types.ListToolsResult(
|
||||
tools=[
|
||||
types.Tool(
|
||||
name="test_tool",
|
||||
description="Test tool",
|
||||
inputSchema={"type": "object", "properties": {"param": {"type": "string"}}},
|
||||
)
|
||||
]
|
||||
)
|
||||
)
|
||||
self.session.call_tool = AsyncMock(
|
||||
return_value=types.CallToolResult(content=[types.TextContent(type="text", text="result")])
|
||||
)
|
||||
|
||||
def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]:
|
||||
return None # type: ignore[return-value] # pyrefly: ignore[bad-return] # ty: ignore[invalid-return-type]
|
||||
|
||||
server = TestServer(name="test_server")
|
||||
async with server:
|
||||
await server.load_tools()
|
||||
|
||||
with (
|
||||
trace.use_span(trace.NonRecordingSpan(trace.INVALID_SPAN_CONTEXT)),
|
||||
patch("agent_framework._mcp.propagate.inject", side_effect=lambda carrier: None),
|
||||
):
|
||||
await server.functions[0].invoke(
|
||||
arguments={"param": "test_value", "_meta": {"attacker.example/route": "evil"}}
|
||||
)
|
||||
|
||||
call_kwargs = server.session.call_tool.call_args.kwargs # type: ignore[union-attr] # ty: ignore[unresolved-attribute]
|
||||
assert call_kwargs["arguments"] == {"param": "test_value"}
|
||||
assert call_kwargs["meta"] is None
|
||||
|
||||
|
||||
async def test_mcp_tool_function_invocation_preserves_trusted_meta_over_model_meta() -> None:
|
||||
"""Trusted function-invocation _meta should be restored after model arguments are merged."""
|
||||
from opentelemetry import trace
|
||||
|
||||
trusted_meta = {"trusted.example/route": "trusted"}
|
||||
|
||||
class TestServer(MCPTool):
|
||||
async def connect(self) -> None: # type: ignore[override] # pyrefly: ignore[bad-override] # ty: ignore[invalid-method-override]
|
||||
self.session = Mock(spec=ClientSession)
|
||||
self.session.list_tools = AsyncMock(
|
||||
return_value=types.ListToolsResult(
|
||||
tools=[
|
||||
types.Tool(
|
||||
name="test_tool",
|
||||
description="Test tool",
|
||||
inputSchema={"type": "object", "properties": {"param": {"type": "string"}}},
|
||||
)
|
||||
]
|
||||
)
|
||||
)
|
||||
self.session.call_tool = AsyncMock(
|
||||
return_value=types.CallToolResult(content=[types.TextContent(type="text", text="result")])
|
||||
)
|
||||
|
||||
def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]:
|
||||
return None # type: ignore[return-value] # pyrefly: ignore[bad-return] # ty: ignore[invalid-return-type]
|
||||
|
||||
server = TestServer(name="test_server")
|
||||
async with server:
|
||||
await server.load_tools()
|
||||
|
||||
context = FunctionInvocationContext(
|
||||
function=server.functions[0],
|
||||
arguments={},
|
||||
kwargs={"_meta": trusted_meta},
|
||||
)
|
||||
with (
|
||||
trace.use_span(trace.NonRecordingSpan(trace.INVALID_SPAN_CONTEXT)),
|
||||
patch("agent_framework._mcp.propagate.inject", side_effect=lambda carrier: None),
|
||||
):
|
||||
await server.functions[0].invoke(
|
||||
arguments={"param": "test_value", "_meta": {"attacker.example/route": "evil"}},
|
||||
context=context,
|
||||
)
|
||||
|
||||
call_kwargs = server.session.call_tool.call_args.kwargs # type: ignore[union-attr] # ty: ignore[unresolved-attribute]
|
||||
assert call_kwargs["arguments"] == {"param": "test_value"}
|
||||
assert call_kwargs["meta"] == trusted_meta
|
||||
|
||||
|
||||
async def test_mcp_tool_call_tool_otel_meta_overrides_user_meta_but_not_tool_list_meta() -> None:
|
||||
"""OpenTelemetry should override caller metadata while tools/list metadata remains most trusted."""
|
||||
from opentelemetry import trace
|
||||
|
||||
tool_meta = {"traceparent": "tool-traceparent", "from_tool": "tool-value"}
|
||||
user_meta = {"traceparent": "user-traceparent", "from_user": "user-value"}
|
||||
|
||||
class TestServer(MCPTool):
|
||||
async def connect(self) -> None: # type: ignore[override] # pyrefly: ignore[bad-override] # ty: ignore[invalid-method-override]
|
||||
self.session = Mock(spec=ClientSession)
|
||||
self.session.list_tools = AsyncMock(
|
||||
return_value=types.ListToolsResult(
|
||||
tools=[
|
||||
types.Tool(
|
||||
name="test_tool",
|
||||
description="Test tool",
|
||||
inputSchema={"type": "object", "properties": {"param": {"type": "string"}}},
|
||||
_meta=tool_meta,
|
||||
)
|
||||
]
|
||||
)
|
||||
)
|
||||
self.session.call_tool = AsyncMock(
|
||||
return_value=types.CallToolResult(content=[types.TextContent(type="text", text="result")])
|
||||
)
|
||||
|
||||
def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]:
|
||||
return None # type: ignore[return-value] # pyrefly: ignore[bad-return] # ty: ignore[invalid-return-type]
|
||||
|
||||
server = TestServer(name="test_server")
|
||||
async with server:
|
||||
await server.load_tools()
|
||||
|
||||
with (
|
||||
trace.use_span(trace.NonRecordingSpan(trace.INVALID_SPAN_CONTEXT)),
|
||||
patch(
|
||||
"agent_framework._mcp.propagate.inject",
|
||||
side_effect=lambda carrier: carrier.update({"traceparent": "otel-traceparent"}),
|
||||
),
|
||||
):
|
||||
await server.call_tool("test_tool", param="test_value", _meta=user_meta)
|
||||
|
||||
call_kwargs = server.session.call_tool.call_args.kwargs # type: ignore[union-attr] # ty: ignore[unresolved-attribute]
|
||||
assert call_kwargs["meta"] == {
|
||||
"traceparent": "tool-traceparent",
|
||||
"from_tool": "tool-value",
|
||||
"from_user": "user-value",
|
||||
}
|
||||
|
||||
|
||||
async def test_mcp_streamable_http_tool_hook_not_duplicated_on_repeated_get_mcp_client():
|
||||
"""Test that calling get_mcp_client multiple times does not accumulate duplicate hooks."""
|
||||
tool = MCPStreamableHTTPTool(
|
||||
@@ -6725,30 +6475,6 @@ def test_prepare_call_kwargs_extracts_meta() -> None:
|
||||
assert meta.get("trace") == "abc"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"key",
|
||||
[
|
||||
"",
|
||||
"_leading-underscore",
|
||||
"trailing-underscore_",
|
||||
"abc/",
|
||||
"1bad.example/name",
|
||||
"bad..example/name",
|
||||
"bad.example/_name",
|
||||
"bad.example/name_",
|
||||
],
|
||||
)
|
||||
def test_prepare_call_kwargs_rejects_invalid_meta_key_names(key: str) -> None:
|
||||
server = MCPTool(name="test_server") # type: ignore[abstract]
|
||||
server._tool_param_names_by_name = {"test_tool": {"param"}}
|
||||
|
||||
with pytest.raises(ToolExecutionException, match="Invalid MCP _meta key name"):
|
||||
server._prepare_call_kwargs(
|
||||
"test_tool",
|
||||
{"param": "v", "_meta": {key: "value"}},
|
||||
)
|
||||
|
||||
|
||||
async def test_call_tool_forwards_only_declared_arguments() -> None:
|
||||
"""End-to-end: framework runtime kwargs are stripped before reaching the server."""
|
||||
|
||||
|
||||
@@ -3860,61 +3860,6 @@ class TestResponseStreamMapAndWithFinalizer:
|
||||
final = await outer.get_final_response()
|
||||
assert final.text == "mapped_update_0mapped_update_1"
|
||||
|
||||
async def test_flat_map_expands_updates(self) -> None:
|
||||
"""flat_map() can transform one update into many updates."""
|
||||
inner = ResponseStream(_generate_updates(2), finalizer=_combine_updates)
|
||||
|
||||
def expand(update: ChatResponseUpdate) -> list[ChatResponseUpdate]:
|
||||
return [
|
||||
ChatResponseUpdate(contents=[Content.from_text(update.text)], role=cast(Any, update.role)),
|
||||
ChatResponseUpdate(contents=[Content.from_text(f"{update.text}_extra")], role=cast(Any, update.role)),
|
||||
]
|
||||
|
||||
outer = inner.flat_map(expand, _combine_updates)
|
||||
|
||||
collected: list[str] = []
|
||||
async for update in outer:
|
||||
collected.append(update.text or "")
|
||||
|
||||
assert collected == ["update_0", "update_0_extra", "update_1", "update_1_extra"]
|
||||
|
||||
final = await outer.get_final_response()
|
||||
assert final.text == "update_0update_0_extraupdate_1update_1_extra"
|
||||
|
||||
async def test_flat_map_skips_empty_mappings(self) -> None:
|
||||
"""flat_map() supports zero-output transforms."""
|
||||
inner = ResponseStream(_generate_updates(3), finalizer=_combine_updates)
|
||||
|
||||
def keep_odd(update: ChatResponseUpdate) -> list[ChatResponseUpdate]:
|
||||
return [update] if update.text == "update_1" else []
|
||||
|
||||
outer = inner.flat_map(keep_odd, _combine_updates)
|
||||
|
||||
collected = [update.text async for update in outer]
|
||||
assert collected == ["update_1"]
|
||||
|
||||
final = await outer.get_final_response()
|
||||
assert final.text == "update_1"
|
||||
|
||||
async def test_flat_map_calls_inner_result_hooks(self) -> None:
|
||||
"""flat_map() preserves inner result hooks."""
|
||||
inner_result_hook_called = {"value": False}
|
||||
|
||||
def inner_result_hook(response: ChatResponse) -> ChatResponse:
|
||||
inner_result_hook_called["value"] = True
|
||||
return response
|
||||
|
||||
inner = ResponseStream(
|
||||
_generate_updates(2),
|
||||
finalizer=_combine_updates,
|
||||
result_hooks=[inner_result_hook], # ty: ignore[invalid-argument-type]
|
||||
)
|
||||
outer = inner.flat_map(lambda u: [u], _combine_updates)
|
||||
|
||||
await outer.get_final_response()
|
||||
|
||||
assert inner_result_hook_called["value"] is True
|
||||
|
||||
async def test_outer_transform_hooks_independent(self) -> None:
|
||||
"""Outer stream has its own independent transform hooks."""
|
||||
inner_hook_calls = {"value": 0}
|
||||
|
||||
@@ -336,97 +336,6 @@ async def test_workflow_checkpoint_chaining_via_previous_checkpoint_id():
|
||||
)
|
||||
|
||||
|
||||
async def test_workflow_checkpoint_ancestry_preserved_after_resume():
|
||||
"""Resuming from a checkpoint must preserve ancestry: future checkpoints chain back to the resumed one."""
|
||||
from typing_extensions import Never
|
||||
|
||||
from agent_framework import WorkflowBuilder, WorkflowContext, handler
|
||||
from agent_framework._workflows._executor import Executor
|
||||
|
||||
class StartExecutor(Executor):
|
||||
@handler
|
||||
async def run(self, message: str, ctx: WorkflowContext[str]) -> None:
|
||||
await ctx.send_message(message, target_id="middle")
|
||||
|
||||
class MiddleExecutor(Executor):
|
||||
@handler
|
||||
async def process(self, message: str, ctx: WorkflowContext[str]) -> None:
|
||||
await ctx.send_message(message + "-processed", target_id="finish")
|
||||
|
||||
class FinishExecutor(Executor):
|
||||
@handler
|
||||
async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None: # type: ignore[valid-type]
|
||||
await ctx.yield_output(message + "-done")
|
||||
|
||||
storage = InMemoryCheckpointStorage()
|
||||
|
||||
def _build_workflow() -> Any:
|
||||
start = StartExecutor(id="start")
|
||||
middle = MiddleExecutor(id="middle")
|
||||
finish = FinishExecutor(id="finish")
|
||||
return (
|
||||
WorkflowBuilder(
|
||||
name="resume-ancestry-test",
|
||||
max_iterations=10,
|
||||
start_executor=start,
|
||||
checkpoint_storage=storage,
|
||||
)
|
||||
.add_edge(start, middle)
|
||||
.add_edge(middle, finish)
|
||||
.build()
|
||||
)
|
||||
|
||||
# First run: produce an initial chain of checkpoints
|
||||
workflow = _build_workflow()
|
||||
workflow_name = workflow.name
|
||||
_ = [event async for event in workflow.run("hello", stream=True)]
|
||||
|
||||
initial_checkpoints = sorted(await storage.list_checkpoints(workflow_name=workflow_name), key=lambda c: c.timestamp)
|
||||
assert len(initial_checkpoints) >= 3, (
|
||||
f"Need at least 3 initial checkpoints to pick a middle one, got {len(initial_checkpoints)}"
|
||||
)
|
||||
initial_ids = {cp.checkpoint_id for cp in initial_checkpoints}
|
||||
|
||||
# Pick an intermediate checkpoint to resume from (not the first, not the last)
|
||||
resume_from = initial_checkpoints[len(initial_checkpoints) // 2]
|
||||
|
||||
# Resume on a fresh workflow instance (same graph signature) and run to completion
|
||||
resumed_workflow = _build_workflow()
|
||||
assert resumed_workflow.name == workflow_name
|
||||
_ = [event async for event in resumed_workflow.run(checkpoint_id=resume_from.checkpoint_id, stream=True)]
|
||||
|
||||
# Inspect new checkpoints created after resuming
|
||||
all_checkpoints = sorted(await storage.list_checkpoints(workflow_name=workflow_name), key=lambda c: c.timestamp)
|
||||
new_checkpoints = [cp for cp in all_checkpoints if cp.checkpoint_id not in initial_ids]
|
||||
assert new_checkpoints, "Resuming from an intermediate checkpoint should produce new checkpoints"
|
||||
|
||||
# The very first checkpoint created after resuming must chain back to the resumed checkpoint
|
||||
assert new_checkpoints[0].previous_checkpoint_id == resume_from.checkpoint_id, (
|
||||
"First post-resume checkpoint must chain to the checkpoint that was resumed from; "
|
||||
f"got previous_checkpoint_id={new_checkpoints[0].previous_checkpoint_id!r}, "
|
||||
f"expected {resume_from.checkpoint_id!r}"
|
||||
)
|
||||
|
||||
# Subsequent post-resume checkpoints must continue chaining
|
||||
for i in range(1, len(new_checkpoints)):
|
||||
assert new_checkpoints[i].previous_checkpoint_id == new_checkpoints[i - 1].checkpoint_id, (
|
||||
f"Post-resume checkpoint {i} should chain to checkpoint {i - 1}"
|
||||
)
|
||||
|
||||
# Walking the chain backwards from the most recent checkpoint must reach the original root
|
||||
# without breaks (i.e. the full ancestry across the resume boundary is intact).
|
||||
checkpoints_by_id = {cp.checkpoint_id: cp for cp in all_checkpoints}
|
||||
chain: list[str] = []
|
||||
cursor: str | None = new_checkpoints[-1].checkpoint_id
|
||||
while cursor is not None:
|
||||
chain.append(cursor)
|
||||
cursor = checkpoints_by_id[cursor].previous_checkpoint_id
|
||||
# Chain must include the resumed-from checkpoint and terminate at the original root
|
||||
assert resume_from.checkpoint_id in chain
|
||||
assert chain[-1] == initial_checkpoints[0].checkpoint_id
|
||||
assert checkpoints_by_id[chain[-1]].previous_checkpoint_id is None
|
||||
|
||||
|
||||
async def test_memory_checkpoint_storage_roundtrip_json_native_types():
|
||||
"""Test that JSON-native types (str, int, float, bool, None) roundtrip correctly."""
|
||||
storage = InMemoryCheckpointStorage()
|
||||
|
||||
@@ -23,7 +23,7 @@ class StartExecutor(Executor):
|
||||
|
||||
class FinishExecutor(Executor):
|
||||
@handler
|
||||
async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None: # zuban: ignore
|
||||
async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None: # type: ignore[valid-type]
|
||||
await ctx.yield_output(message)
|
||||
|
||||
|
||||
@@ -95,7 +95,7 @@ class SubStartExecutor(Executor):
|
||||
|
||||
class SubFinishExecutor(Executor):
|
||||
@handler
|
||||
async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None: # zuban: ignore
|
||||
async def finish(self, message: str, ctx: WorkflowContext[Never, str]) -> None: # type: ignore[valid-type]
|
||||
await ctx.yield_output(message)
|
||||
|
||||
|
||||
|
||||
@@ -17,6 +17,7 @@ from agent_framework import (
|
||||
WorkflowContext,
|
||||
WorkflowConvergenceException,
|
||||
WorkflowEvent,
|
||||
WorkflowRunnerException,
|
||||
WorkflowRunState,
|
||||
handler,
|
||||
)
|
||||
@@ -304,62 +305,40 @@ async def test_fanout_edge_runner_delivers_to_multiple_targets_concurrently() ->
|
||||
assert probe_target.call_count == 1
|
||||
|
||||
|
||||
async def test_runner_run_until_convergence_runs_sequentially():
|
||||
"""run_until_convergence can be invoked back-to-back on the same Runner.
|
||||
|
||||
The Runner itself does not enforce concurrency; that responsibility lives on
|
||||
:class:`Workflow`. This test simply confirms the Runner is reusable across
|
||||
sequential runs.
|
||||
"""
|
||||
runner = _make_runner()
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
|
||||
|
||||
def _make_runner() -> Runner:
|
||||
"""Build a minimal runner for runner-level tests."""
|
||||
return Runner(
|
||||
[],
|
||||
{},
|
||||
State(),
|
||||
InProcRunnerContext(),
|
||||
"test_name",
|
||||
graph_signature_hash="test_hash",
|
||||
)
|
||||
|
||||
|
||||
async def test_runner_accepts_new_run_after_previous_failure():
|
||||
"""A failed run must not leave the Runner unable to start a new run.
|
||||
|
||||
After the first run raises, ``run_until_convergence()`` must be callable
|
||||
again and not surface any lifecycle-related rejection.
|
||||
"""
|
||||
async def test_runner_already_running():
|
||||
"""Test that running the runner while it is already running raises an error."""
|
||||
executor_a = MockExecutor(id="executor_a")
|
||||
executor_b = MockExecutor(id="executor_b")
|
||||
|
||||
# Create a loop
|
||||
edges = [
|
||||
SingleEdgeGroup(executor_a.id, executor_b.id),
|
||||
SingleEdgeGroup(executor_b.id, executor_a.id),
|
||||
]
|
||||
executors: dict[str, Executor] = {executor_a.id: executor_a, executor_b.id: executor_b}
|
||||
|
||||
executors: dict[str, Executor] = {
|
||||
executor_a.id: executor_a,
|
||||
executor_b.id: executor_b,
|
||||
}
|
||||
state = State()
|
||||
ctx = InProcRunnerContext()
|
||||
runner = Runner(edges, executors, state, ctx, "test_name", graph_signature_hash="test_hash", max_iterations=2)
|
||||
|
||||
await executor_a.execute(MockMessage(data=0), ["START"], state, ctx)
|
||||
runner = Runner(edges, executors, state, ctx, "test_name", graph_signature_hash="test_hash")
|
||||
|
||||
with pytest.raises(WorkflowConvergenceException):
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
await executor_a.execute(
|
||||
MockMessage(data=0),
|
||||
["START"], # source_executor_ids
|
||||
state, # state
|
||||
ctx, # runner_context
|
||||
)
|
||||
|
||||
# A second run on the same Runner must not be blocked by stale lifecycle
|
||||
# state from the failed run.
|
||||
try:
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
except Exception as exc:
|
||||
assert "Runner is already running" not in str(exc), "Runner stayed locked after a failed run"
|
||||
with pytest.raises(WorkflowRunnerException, match="Runner is already running."):
|
||||
|
||||
async def _run():
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
|
||||
await asyncio.gather(_run(), _run())
|
||||
|
||||
|
||||
async def test_runner_emits_runner_completion_for_agent_response_without_targets():
|
||||
@@ -883,13 +862,7 @@ async def test_runner_checkpoint_with_resumed_flag():
|
||||
state = State()
|
||||
|
||||
runner = Runner(edges, executors, state, ctx, "test_name", graph_signature_hash="test_hash")
|
||||
resumed_checkpoint = WorkflowCheckpoint(
|
||||
checkpoint_id="resumed-cp",
|
||||
workflow_name="test_name",
|
||||
graph_signature_hash="test_hash",
|
||||
iteration_count=5,
|
||||
)
|
||||
runner._mark_resumed(resumed_checkpoint) # pyright: ignore[reportPrivateUsage]
|
||||
runner._mark_resumed(5) # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
# Add a message to trigger the checkpoint creation path
|
||||
await ctx.send_message(WorkflowMessage(data=MockMessage(data=8), source_id="START"))
|
||||
@@ -909,86 +882,6 @@ async def test_runner_checkpoint_with_resumed_flag():
|
||||
assert runner._resumed_from_checkpoint is False # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
|
||||
async def test_runner_mark_resumed_sets_previous_checkpoint_id():
|
||||
"""_mark_resumed must populate _previous_checkpoint_id so future checkpoints chain back to the resume point."""
|
||||
runner = Runner(
|
||||
[],
|
||||
{},
|
||||
State(),
|
||||
InProcRunnerContext(),
|
||||
"test_name",
|
||||
graph_signature_hash="test_hash",
|
||||
)
|
||||
|
||||
# Pre-condition: nothing to chain back to
|
||||
assert runner._previous_checkpoint_id is None # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
resumed_checkpoint = WorkflowCheckpoint(
|
||||
checkpoint_id="resumed-cp-id",
|
||||
workflow_name="test_name",
|
||||
graph_signature_hash="test_hash",
|
||||
iteration_count=3,
|
||||
)
|
||||
runner._mark_resumed(resumed_checkpoint) # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
assert runner._resumed_from_checkpoint is True # pyright: ignore[reportPrivateUsage]
|
||||
assert runner._iteration == 3 # pyright: ignore[reportPrivateUsage]
|
||||
assert runner._previous_checkpoint_id == "resumed-cp-id" # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
|
||||
async def test_runner_post_resume_checkpoint_chains_to_resumed_checkpoint():
|
||||
"""After resuming, the next checkpoint created must reference the resumed checkpoint as its parent."""
|
||||
storage = InMemoryCheckpointStorage()
|
||||
ctx = CheckpointingContext(storage)
|
||||
executor_a = MockExecutor(id="executor_a")
|
||||
executor_b = MockExecutor(id="executor_b")
|
||||
|
||||
edges = [
|
||||
SingleEdgeGroup(executor_a.id, executor_b.id),
|
||||
SingleEdgeGroup(executor_b.id, executor_a.id),
|
||||
]
|
||||
|
||||
executors: dict[str, Executor] = {
|
||||
executor_a.id: executor_a,
|
||||
executor_b.id: executor_b,
|
||||
}
|
||||
state = State()
|
||||
|
||||
runner = Runner(edges, executors, state, ctx, "test_name", graph_signature_hash="test_hash")
|
||||
|
||||
# Simulate having resumed from a prior checkpoint
|
||||
resumed_checkpoint = WorkflowCheckpoint(
|
||||
checkpoint_id="parent-checkpoint-id",
|
||||
workflow_name="test_name",
|
||||
graph_signature_hash="test_hash",
|
||||
iteration_count=1,
|
||||
)
|
||||
runner._mark_resumed(resumed_checkpoint) # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
# Seed a message so the runner has work to do (and creates checkpoints at superstep boundaries)
|
||||
await ctx.send_message(WorkflowMessage(data=MockMessage(data=8), source_id=executor_a.id))
|
||||
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
|
||||
# Find the first checkpoint created after the resume point (across all workflows tracked by storage)
|
||||
new_checkpoints = sorted(
|
||||
await storage.list_checkpoints(workflow_name="test_name"),
|
||||
key=lambda c: c.timestamp,
|
||||
)
|
||||
assert new_checkpoints, "Resuming and running should produce at least one new checkpoint"
|
||||
|
||||
# The first new checkpoint must chain to the resumed-from checkpoint, not to None
|
||||
assert new_checkpoints[0].previous_checkpoint_id == "parent-checkpoint-id", (
|
||||
"First post-resume checkpoint must chain to the resumed checkpoint id; "
|
||||
f"got {new_checkpoints[0].previous_checkpoint_id!r}"
|
||||
)
|
||||
|
||||
# Subsequent post-resume checkpoints continue the chain
|
||||
for i in range(1, len(new_checkpoints)):
|
||||
assert new_checkpoints[i].previous_checkpoint_id == new_checkpoints[i - 1].checkpoint_id
|
||||
|
||||
|
||||
class ExecutorThatFailsWithEvents(Executor):
|
||||
"""An executor that emits events and then raises an exception after receiving messages."""
|
||||
|
||||
@@ -1058,172 +951,6 @@ async def test_runner_drains_events_on_iteration_exception():
|
||||
assert len(output_events) >= 1
|
||||
|
||||
|
||||
async def test_runner_resumed_flag_reset_after_failed_resumed_run():
|
||||
"""A failed *resumed* run must not leak the resume flag into the next run.
|
||||
|
||||
The resume flag suppresses the initial "superstep 0" (entry) checkpoint when resuming from an
|
||||
iteration-0 checkpoint (which already exists and must not be recreated). It used to be cleared
|
||||
only on the success path, so an executor failure during a resumed run left it ``True`` and the
|
||||
next fresh run wrongly skipped its entry checkpoint. The flag is now cleared in a ``finally`` so
|
||||
this holds even when convergence raises.
|
||||
|
||||
This also verifies checkpoint creation on the re-run: the resumed (failed) run creates no entry
|
||||
checkpoint, while the subsequent fresh run does.
|
||||
"""
|
||||
storage = InMemoryCheckpointStorage()
|
||||
ctx = CheckpointingContext(storage)
|
||||
executor_a = PassthroughExecutor(id="executor_a")
|
||||
executor_b = ExecutorThatFailsWithEvents(id="executor_b", runner_ctx=ctx, fail_on_iteration=1)
|
||||
|
||||
edges = [SingleEdgeGroup(executor_a.id, executor_b.id)]
|
||||
executors: dict[str, Executor] = {executor_a.id: executor_a, executor_b.id: executor_b}
|
||||
state = State()
|
||||
|
||||
runner = Runner(edges, executors, state, ctx, "test_name", graph_signature_hash="test_hash")
|
||||
|
||||
# Simulate a resumed run; this marks the runner as resumed so the next run skips
|
||||
# the superstep-0 checkpoint.
|
||||
resumed_checkpoint = WorkflowCheckpoint(
|
||||
checkpoint_id="resumed-cp",
|
||||
workflow_name="test_name",
|
||||
graph_signature_hash="test_hash",
|
||||
iteration_count=0,
|
||||
)
|
||||
runner._mark_resumed(resumed_checkpoint) # pyright: ignore[reportPrivateUsage]
|
||||
assert runner._resumed_from_checkpoint is True # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
# Run the resumed turn; executor_b fails mid-iteration before any superstep
|
||||
# checkpoint is created.
|
||||
await executor_a.execute(MockMessage(data=0), ["START"], state, ctx)
|
||||
with pytest.raises(RuntimeError, match="Executor failed with pending events"):
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
|
||||
# The fix: the resume flag is cleared even though the run raised.
|
||||
assert runner._resumed_from_checkpoint is False # pyright: ignore[reportPrivateUsage]
|
||||
# The resumed (failed) run created no superstep-0 checkpoint (it was skipped).
|
||||
assert await storage.list_checkpoints(workflow_name="test_name") == []
|
||||
|
||||
# Re-run as a fresh turn: with the flag correctly reset, the runner now creates
|
||||
# the initial superstep-0 checkpoint (iteration_count == 0) before failing again.
|
||||
runner.reset_iteration_count()
|
||||
await executor_a.execute(MockMessage(data=0), ["START"], state, ctx)
|
||||
with pytest.raises(RuntimeError, match="Executor failed with pending events"):
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
|
||||
checkpoints = await storage.list_checkpoints(workflow_name="test_name")
|
||||
assert any(cp.iteration_count == 0 for cp in checkpoints), (
|
||||
"Fresh run after a failed resumed run must create the superstep-0 checkpoint; "
|
||||
"a leaked resume flag would have skipped it"
|
||||
)
|
||||
|
||||
|
||||
async def test_runner_creates_entry_checkpoint_at_iteration_zero():
|
||||
"""A fresh run creates the entry (superstep-0) checkpoint at iteration 0 with no parent.
|
||||
|
||||
This is the baseline the lineage-consistency guard must preserve: when starting from iteration 0
|
||||
with messages queued and not resumed, the entry checkpoint is created and begins a new lineage
|
||||
(``previous_checkpoint_id is None``).
|
||||
"""
|
||||
storage = InMemoryCheckpointStorage()
|
||||
ctx = CheckpointingContext(storage)
|
||||
# Terminal executor with no outgoing edges: the runner runs one superstep and converges.
|
||||
source = MockExecutor(id="source")
|
||||
state = State()
|
||||
runner = Runner([], {source.id: source}, state, ctx, "test_name", graph_signature_hash="test_hash")
|
||||
|
||||
assert runner._iteration == 0 # pyright: ignore[reportPrivateUsage]
|
||||
assert runner._resumed_from_checkpoint is False # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
await ctx.send_message(WorkflowMessage(data=MockMessage(data=8), source_id=source.id))
|
||||
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
|
||||
checkpoints = await storage.list_checkpoints(workflow_name="test_name")
|
||||
entry_checkpoints = [cp for cp in checkpoints if cp.iteration_count == 0]
|
||||
assert len(entry_checkpoints) == 1, "A fresh run must create exactly one entry checkpoint at iteration 0"
|
||||
assert entry_checkpoints[0].previous_checkpoint_id is None, (
|
||||
"The entry checkpoint of a fresh run must begin a new lineage with no parent"
|
||||
)
|
||||
|
||||
|
||||
async def test_runner_skips_entry_checkpoint_when_iteration_nonzero():
|
||||
"""The entry (superstep-0) checkpoint must only be created at iteration 0 to keep lineage consistent.
|
||||
|
||||
A re-run that did not reset the iteration count (and is not marked as resumed) must not write an
|
||||
entry checkpoint carrying a non-zero ``iteration_count`` - doing so would place two checkpoints at
|
||||
the same iteration in the lineage. The ``_iteration == 0`` guard suppresses the entry checkpoint in
|
||||
this case while still allowing the normal per-superstep checkpoints to be created.
|
||||
"""
|
||||
storage = InMemoryCheckpointStorage()
|
||||
ctx = CheckpointingContext(storage)
|
||||
# Terminal executor with no outgoing edges: the runner runs one superstep and converges.
|
||||
source = MockExecutor(id="source")
|
||||
state = State()
|
||||
runner = Runner([], {source.id: source}, state, ctx, "test_name", graph_signature_hash="test_hash")
|
||||
|
||||
# Simulate a re-run that kept its iteration count and is not marked as resumed.
|
||||
runner._iteration = 5 # pyright: ignore[reportPrivateUsage]
|
||||
assert runner._resumed_from_checkpoint is False # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
await ctx.send_message(WorkflowMessage(data=MockMessage(data=8), source_id=source.id))
|
||||
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
|
||||
checkpoints = await storage.list_checkpoints(workflow_name="test_name")
|
||||
# No entry checkpoint at the pre-existing iteration count may be created.
|
||||
assert all(cp.iteration_count != 5 for cp in checkpoints), (
|
||||
"Entry checkpoint must not be created at a non-zero iteration; lineage would have a duplicate iteration"
|
||||
)
|
||||
# The normal post-superstep checkpoint is still created (iteration advanced to 6).
|
||||
assert any(cp.iteration_count == 6 for cp in checkpoints)
|
||||
|
||||
|
||||
async def test_runner_resumed_from_iteration_zero_skips_entry_checkpoint():
|
||||
"""Resuming from an iteration-0 checkpoint must not recreate the entry checkpoint.
|
||||
|
||||
Here ``_iteration == 0`` is true, so the iteration guard alone would not suppress the entry
|
||||
checkpoint; the resume flag is what prevents recreating the checkpoint that already exists at
|
||||
iteration 0.
|
||||
"""
|
||||
storage = InMemoryCheckpointStorage()
|
||||
ctx = CheckpointingContext(storage)
|
||||
source = MockExecutor(id="source")
|
||||
state = State()
|
||||
runner = Runner([], {source.id: source}, state, ctx, "test_name", graph_signature_hash="test_hash")
|
||||
|
||||
# Resume from an iteration-0 checkpoint: iteration stays 0 but the run is marked as resumed.
|
||||
resumed_checkpoint = WorkflowCheckpoint(
|
||||
checkpoint_id="entry-cp",
|
||||
workflow_name="test_name",
|
||||
graph_signature_hash="test_hash",
|
||||
iteration_count=0,
|
||||
)
|
||||
runner._mark_resumed(resumed_checkpoint) # pyright: ignore[reportPrivateUsage]
|
||||
assert runner._iteration == 0 # pyright: ignore[reportPrivateUsage]
|
||||
assert runner._resumed_from_checkpoint is True # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
await ctx.send_message(WorkflowMessage(data=MockMessage(data=8), source_id=source.id))
|
||||
|
||||
async for _ in runner.run_until_convergence():
|
||||
pass
|
||||
|
||||
# The pre-loop entry checkpoint is skipped; only the post-superstep checkpoint (iteration 1) is created,
|
||||
# and it chains back to the resumed entry checkpoint.
|
||||
checkpoints = sorted(
|
||||
await storage.list_checkpoints(workflow_name="test_name"),
|
||||
key=lambda c: c.timestamp,
|
||||
)
|
||||
assert all(cp.checkpoint_id != "entry-cp" for cp in checkpoints), "Resumed entry checkpoint must not be recreated"
|
||||
assert checkpoints, "The resumed run must still create its post-superstep checkpoint"
|
||||
assert checkpoints[0].previous_checkpoint_id == "entry-cp", (
|
||||
"The first post-resume checkpoint must chain back to the resumed entry checkpoint"
|
||||
)
|
||||
|
||||
|
||||
class SlowEventEmittingExecutor(Executor):
|
||||
"""An executor that emits events with delays to test straggler event draining."""
|
||||
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import gc
|
||||
import tempfile
|
||||
from collections.abc import AsyncIterable, Awaitable, Sequence
|
||||
from dataclasses import dataclass, field
|
||||
@@ -20,7 +19,6 @@ from agent_framework import (
|
||||
Content,
|
||||
Executor,
|
||||
FileCheckpointStorage,
|
||||
InProcRunnerContext,
|
||||
Message,
|
||||
ResponseStream,
|
||||
WorkflowBuilder,
|
||||
@@ -28,7 +26,6 @@ from agent_framework import (
|
||||
WorkflowContext,
|
||||
WorkflowConvergenceException,
|
||||
WorkflowEvent,
|
||||
WorkflowException,
|
||||
WorkflowMessage,
|
||||
WorkflowRunState,
|
||||
handler,
|
||||
@@ -762,7 +759,8 @@ async def test_workflow_concurrent_execution_prevention():
|
||||
|
||||
# Try to start a second concurrent execution - this should fail
|
||||
with pytest.raises(
|
||||
WorkflowException, match="Workflow is already running; concurrent runs are not allowed on the same instance."
|
||||
RuntimeError,
|
||||
match="Workflow is already running. Concurrent executions are not allowed.",
|
||||
):
|
||||
await workflow.run(NumberMessage(data=0))
|
||||
|
||||
@@ -797,7 +795,8 @@ async def test_workflow_concurrent_execution_prevention_streaming():
|
||||
|
||||
# Try to start a second concurrent execution - this should fail
|
||||
with pytest.raises(
|
||||
WorkflowException, match="Workflow is already running; concurrent runs are not allowed on the same instance."
|
||||
RuntimeError,
|
||||
match="Workflow is already running. Concurrent executions are not allowed.",
|
||||
):
|
||||
await workflow.run(NumberMessage(data=0))
|
||||
|
||||
@@ -829,12 +828,14 @@ async def test_workflow_concurrent_execution_prevention_mixed_methods():
|
||||
|
||||
# Try different execution methods - all should fail
|
||||
with pytest.raises(
|
||||
WorkflowException, match="Workflow is already running; concurrent runs are not allowed on the same instance."
|
||||
RuntimeError,
|
||||
match="Workflow is already running. Concurrent executions are not allowed.",
|
||||
):
|
||||
await workflow.run(NumberMessage(data=0))
|
||||
|
||||
with pytest.raises(
|
||||
WorkflowException, match="Workflow is already running; concurrent runs are not allowed on the same instance."
|
||||
RuntimeError,
|
||||
match="Workflow is already running. Concurrent executions are not allowed.",
|
||||
):
|
||||
async for _ in workflow.run(NumberMessage(data=0), stream=True):
|
||||
break
|
||||
@@ -847,238 +848,6 @@ async def test_workflow_concurrent_execution_prevention_mixed_methods():
|
||||
assert result.get_final_state() == WorkflowRunState.IDLE
|
||||
|
||||
|
||||
async def test_workflow_sequential_runs_after_completion() -> None:
|
||||
"""A completed run must release the runner so the next ``run`` succeeds.
|
||||
|
||||
This is the happy-path counterpart to the concurrent-run guard tests:
|
||||
those tests verify that a *concurrent* run is rejected, but they do not
|
||||
verify that the lock is actually released afterwards. This test
|
||||
exercises that release path explicitly across the three call shapes
|
||||
(non-streaming, streaming-iterated, streaming-via-get_final_response)
|
||||
and across multiple consecutive turns to catch lock leaks.
|
||||
"""
|
||||
executor = IncrementExecutor(id="seq_executor", limit=3, increment=1)
|
||||
workflow = WorkflowBuilder(start_executor=executor).build()
|
||||
|
||||
# Non-streaming -> non-streaming
|
||||
r1 = await workflow.run(NumberMessage(data=0))
|
||||
assert r1.get_final_state() == WorkflowRunState.IDLE
|
||||
|
||||
r2 = await workflow.run(NumberMessage(data=0))
|
||||
assert r2.get_final_state() == WorkflowRunState.IDLE
|
||||
|
||||
# Non-streaming -> streaming-iterated
|
||||
stream_events: list[WorkflowEvent] = []
|
||||
async for event in workflow.run(NumberMessage(data=0), stream=True):
|
||||
stream_events.append(event)
|
||||
assert any(e.type == "status" and e.state == WorkflowRunState.IDLE for e in stream_events)
|
||||
|
||||
# Streaming -> streaming via get_final_response (no manual iteration)
|
||||
r3 = await workflow.run(NumberMessage(data=0), stream=True).get_final_response()
|
||||
assert r3.get_final_state() == WorkflowRunState.IDLE
|
||||
|
||||
# Streaming -> non-streaming (back to the start)
|
||||
r4 = await workflow.run(NumberMessage(data=0))
|
||||
assert r4.get_final_state() == WorkflowRunState.IDLE
|
||||
|
||||
|
||||
async def test_workflow_unconsumed_stream_releases_run_lock() -> None:
|
||||
"""An unconsumed stream must not leak the run lock.
|
||||
|
||||
``Workflow.run`` reserves the runner *synchronously* so that concurrent
|
||||
callers are rejected immediately. The reservation is normally released
|
||||
by ``_run_core``'s ``finally`` once the stream is iterated. If the
|
||||
caller never iterates the stream, a GC-time finalizer must release the
|
||||
reservation instead - otherwise every subsequent ``Workflow.run`` call
|
||||
on this instance would fail with the concurrent-run error.
|
||||
"""
|
||||
executor = IncrementExecutor(id="unconsumed_stream_exec", limit=3, increment=1)
|
||||
workflow = WorkflowBuilder(start_executor=executor).build()
|
||||
|
||||
# Build a stream and immediately drop it without iterating.
|
||||
stream = workflow.run(NumberMessage(data=0), stream=True)
|
||||
assert stream is not None # silence unused-variable warnings; stream is GC'd below
|
||||
del stream
|
||||
gc.collect()
|
||||
# Yield to the event loop so any scheduled finalizer work can run.
|
||||
await asyncio.sleep(0)
|
||||
|
||||
# The runner should be back to IDLE; a fresh run must succeed.
|
||||
result = await workflow.run(NumberMessage(data=0))
|
||||
assert result.get_final_state() == WorkflowRunState.IDLE
|
||||
|
||||
|
||||
async def test_workflow_unawaited_run_coroutine_releases_run_lock() -> None:
|
||||
"""An un-awaited non-streaming ``run()`` coroutine must also not leak the lock.
|
||||
|
||||
``Workflow.run`` (non-streaming) returns a coroutine produced by
|
||||
``ResponseStream.get_final_response``. The underlying ResponseStream is
|
||||
held alive by that coroutine, so dropping the coroutine without
|
||||
awaiting it must still release the reservation via the same GC-time
|
||||
fallback used for unconsumed streams.
|
||||
"""
|
||||
executor = IncrementExecutor(id="unawaited_run_exec", limit=3, increment=1)
|
||||
workflow = WorkflowBuilder(start_executor=executor).build()
|
||||
|
||||
coro = workflow.run(NumberMessage(data=0))
|
||||
# Closing suppresses the "coroutine was never awaited" warning. We cast to
|
||||
# ``Any`` because the typed return is ``Awaitable[...]``; in practice it is
|
||||
# a coroutine that exposes ``close``.
|
||||
cast(Any, coro).close()
|
||||
del coro
|
||||
gc.collect()
|
||||
await asyncio.sleep(0)
|
||||
|
||||
result = await workflow.run(NumberMessage(data=0))
|
||||
assert result.get_final_state() == WorkflowRunState.IDLE
|
||||
|
||||
|
||||
async def test_workflow_partial_stream_does_not_clobber_successor_active_run() -> None:
|
||||
"""A stale ``_run_core`` finalizer must not clear a successor's run lock.
|
||||
|
||||
Repro for the GC-finalizer race the user reported:
|
||||
|
||||
1. Start stream A and consume one event so its body is suspended at a
|
||||
``yield``. Its ``finally`` is now armed and will run when the
|
||||
generator is closed.
|
||||
2. Drop stream A and ``gc.collect``. The ``_active_run`` weakref's
|
||||
referent is gone, so a subsequent ``run()`` will pass the
|
||||
concurrency guard - but stream A's async-gen finalizer hasn't
|
||||
actually executed yet (``aclose`` is scheduled on the loop).
|
||||
3. Synchronously start stream B; ``run()`` installs a fresh weakref
|
||||
in ``_active_run``.
|
||||
4. Yield to the loop so stream A's stale ``finally`` runs. Without
|
||||
the identity check it unconditionally writes
|
||||
``self._active_run = None``, silently disabling the concurrency
|
||||
guard for stream B.
|
||||
"""
|
||||
executor = IncrementExecutor(id="stale_finalizer_exec", limit=100, increment=1)
|
||||
workflow = WorkflowBuilder(start_executor=executor).build()
|
||||
|
||||
# Step 1: drive stream A's body until it's suspended at its first yield.
|
||||
stream_a = workflow.run(NumberMessage(data=0), stream=True)
|
||||
aiter_a = stream_a.__aiter__()
|
||||
await aiter_a.__anext__()
|
||||
|
||||
# Step 2: drop stream A; GC invalidates the weakref and schedules
|
||||
# async-gen close, but does not run the close inline.
|
||||
del stream_a
|
||||
del aiter_a
|
||||
gc.collect()
|
||||
|
||||
# Step 3: synchronously start stream B *before* yielding to the loop,
|
||||
# so the stale ``aclose`` for stream A hasn't fired yet.
|
||||
stream_b = workflow.run(NumberMessage(data=0), stream=True)
|
||||
ref_b = workflow._active_run # type: ignore[attr-defined]
|
||||
assert ref_b is not None and ref_b() is stream_b
|
||||
|
||||
# Step 4: yield enough times for stream A's scheduled aclose to drive
|
||||
# its body through ``GeneratorExit`` and into its ``finally``.
|
||||
for _ in range(5):
|
||||
await asyncio.sleep(0)
|
||||
|
||||
# With the fix, stream B's reservation is still in place. Without it,
|
||||
# ``_active_run`` was clobbered to ``None`` and a concurrent run would
|
||||
# be (incorrectly) accepted.
|
||||
assert workflow._active_run is ref_b # type: ignore[attr-defined]
|
||||
with pytest.raises(
|
||||
WorkflowException,
|
||||
match="Workflow is already running; concurrent runs are not allowed on the same instance.",
|
||||
):
|
||||
await workflow.run(NumberMessage(data=0))
|
||||
|
||||
# Tear down stream B without iterating it (its body never started, so
|
||||
# closing it is a no-op for workflow state).
|
||||
del stream_b
|
||||
del ref_b
|
||||
gc.collect()
|
||||
await asyncio.sleep(0)
|
||||
|
||||
|
||||
async def test_workflow_stale_runtime_checkpoint_storage_not_inherited() -> None:
|
||||
"""A new run must not inherit a prior run's leftover runtime checkpoint storage.
|
||||
|
||||
If a run that set a runtime ``checkpoint_storage`` override is dropped before
|
||||
its async-generator finalizer clears it, the override can linger on the
|
||||
``RunnerContext`` while ``_is_run_active()`` already reports False. ``run()``
|
||||
defensively clears that stale override so a subsequent run that does not pass
|
||||
its own ``checkpoint_storage`` does not silently checkpoint into it.
|
||||
"""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
leftover_storage = FileCheckpointStorage(temp_dir)
|
||||
executor = IncrementExecutor(id="stale_storage_exec", limit=3, increment=1)
|
||||
workflow = WorkflowBuilder(start_executor=executor).build()
|
||||
|
||||
assert isinstance(workflow._runner.context, InProcRunnerContext) # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
# Simulate a leftover runtime override from a dropped prior run.
|
||||
workflow._runner.context.set_runtime_checkpoint_storage(leftover_storage) # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
# A fresh run without its own checkpoint_storage must not use the leftover.
|
||||
result = await workflow.run(NumberMessage(data=0))
|
||||
assert result.get_final_state() == WorkflowRunState.IDLE
|
||||
|
||||
checkpoints = await leftover_storage.list_checkpoints(workflow_name=workflow.name)
|
||||
assert checkpoints == [], "Stale runtime checkpoint storage must not be inherited by a new run"
|
||||
assert workflow._runner.context._runtime_checkpoint_storage is None # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
|
||||
async def test_workflow_partial_stream_does_not_clobber_successor_runtime_storage() -> None:
|
||||
"""A stale ``_run_core`` finalizer must not clear a successor's runtime storage.
|
||||
|
||||
Same GC-finalizer race as
|
||||
``test_workflow_partial_stream_does_not_clobber_successor_active_run`` but for the
|
||||
runtime checkpoint storage override: the dropped run's deferred ``finally`` must
|
||||
only clear the override if it still owns it, otherwise it wipes the successor
|
||||
run's storage.
|
||||
"""
|
||||
with (
|
||||
tempfile.TemporaryDirectory() as temp_dir_a,
|
||||
tempfile.TemporaryDirectory() as temp_dir_b,
|
||||
):
|
||||
storage_a = FileCheckpointStorage(temp_dir_a)
|
||||
storage_b = FileCheckpointStorage(temp_dir_b)
|
||||
executor = IncrementExecutor(id="storage_finalizer_exec", limit=100, increment=1)
|
||||
workflow = WorkflowBuilder(start_executor=executor).build()
|
||||
context = workflow._runner.context # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
assert isinstance(context, InProcRunnerContext)
|
||||
|
||||
# Step 1: drive stream A's body to its first yield so it set storage_a.
|
||||
stream_a = workflow.run(NumberMessage(data=0), checkpoint_storage=storage_a, stream=True)
|
||||
aiter_a = stream_a.__aiter__()
|
||||
await aiter_a.__anext__()
|
||||
assert context._runtime_checkpoint_storage is storage_a # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
# Step 2: drop stream A; the weakref dies and async-gen close is scheduled
|
||||
# but not run inline.
|
||||
del stream_a
|
||||
del aiter_a
|
||||
gc.collect()
|
||||
|
||||
# Step 3: synchronously start stream B with its own storage and drive it to
|
||||
# its first yield so it set storage_b and took ownership of the override.
|
||||
stream_b = workflow.run(NumberMessage(data=0), checkpoint_storage=storage_b, stream=True)
|
||||
aiter_b = stream_b.__aiter__()
|
||||
await aiter_b.__anext__()
|
||||
assert context._runtime_checkpoint_storage is storage_b # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
# Step 4: yield enough for stream A's scheduled aclose to drive its body
|
||||
# through ``GeneratorExit`` and into its ``finally``.
|
||||
for _ in range(5):
|
||||
await asyncio.sleep(0)
|
||||
|
||||
# With the ownership guard, stream B's override survives. Without it, A's
|
||||
# stale finalizer would have cleared it.
|
||||
assert context._runtime_checkpoint_storage is storage_b # pyright: ignore[reportPrivateUsage]
|
||||
|
||||
# Tear down stream B.
|
||||
del stream_b
|
||||
del aiter_b
|
||||
gc.collect()
|
||||
await asyncio.sleep(0)
|
||||
|
||||
|
||||
class _StreamingTestAgent(BaseAgent):
|
||||
"""Test agent that supports both streaming and non-streaming modes."""
|
||||
|
||||
|
||||
@@ -90,7 +90,7 @@ async def _run(yaml_def: dict[str, Any], handler: HttpRequestHandler) -> Any:
|
||||
|
||||
def _state(workflow: Any, events: Any) -> dict[str, Any]:
|
||||
"""Read declarative state out of the workflow after run completes."""
|
||||
return workflow._runner.state.get(DECLARATIVE_STATE_KEY) or {}
|
||||
return workflow._state.get(DECLARATIVE_STATE_KEY) or {}
|
||||
|
||||
|
||||
# Helper used by parametrised path tests
|
||||
@@ -151,7 +151,7 @@ class TestSuccessPath:
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(method="GET", response="Local.Result")))
|
||||
await workflow.run({})
|
||||
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["Result"] == {"key": "value", "number": 42}
|
||||
assert handler.last_info is not None
|
||||
assert handler.last_info.method == "GET"
|
||||
@@ -164,7 +164,7 @@ class TestSuccessPath:
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(response="Local.Result")))
|
||||
await workflow.run({})
|
||||
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["Result"] == "not-json content"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -174,7 +174,7 @@ class TestSuccessPath:
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(response="Local.Result")))
|
||||
await workflow.run({})
|
||||
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["Result"] is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -184,7 +184,7 @@ class TestSuccessPath:
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(response={"path": "Local.Result"})))
|
||||
await workflow.run({})
|
||||
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["Result"] == {"x": 1}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -517,7 +517,7 @@ class TestResponseHeaders:
|
||||
factory = WorkflowFactory(http_request_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(response_headers="Local.H")))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
h = decl["Local"]["H"]
|
||||
assert h["Content-Type"] == "application/json"
|
||||
assert h["Set-Cookie"] == "a=1,b=2"
|
||||
@@ -528,7 +528,7 @@ class TestResponseHeaders:
|
||||
factory = WorkflowFactory(http_request_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(response_headers="Local.H")))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["H"] is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -538,7 +538,7 @@ class TestResponseHeaders:
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(response_headers="Local.H")))
|
||||
with pytest.raises(DeclarativeActionError):
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["H"] == {"X-Trace": "abc"}
|
||||
|
||||
|
||||
@@ -559,7 +559,7 @@ class TestConversationAppend:
|
||||
)
|
||||
)
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
conv = decl["System"]["conversations"].get("conv-test-1")
|
||||
assert conv is not None
|
||||
assert len(conv["messages"]) == 1
|
||||
@@ -570,7 +570,7 @@ class TestConversationAppend:
|
||||
factory = WorkflowFactory(http_request_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(response="Local.Result", conversation_id="")))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
# Auto-init creates an entry for the System.ConversationId conversation,
|
||||
# but it should NOT have HTTP-appended messages from us.
|
||||
for _cid, conv in decl["System"]["conversations"].items():
|
||||
@@ -582,7 +582,7 @@ class TestConversationAppend:
|
||||
factory = WorkflowFactory(http_request_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(conversation_id="conv-test-1")))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
# No conversation entry should have been created either.
|
||||
assert "conv-test-1" not in decl["System"]["conversations"]
|
||||
|
||||
|
||||
@@ -73,7 +73,7 @@ async def test_http_request_yaml_roundtrip() -> None:
|
||||
workflow = factory.create_workflow_from_yaml_path(FIXTURE_PATH)
|
||||
await workflow.run({})
|
||||
|
||||
decl: dict[str, Any] = workflow._runner.state.get(DECLARATIVE_STATE_KEY) or {}
|
||||
decl: dict[str, Any] = workflow._state.get(DECLARATIVE_STATE_KEY) or {}
|
||||
local: dict[str, Any] = decl.get("Local") or {}
|
||||
|
||||
assert local.get("RepoOwner") == "dotnet"
|
||||
|
||||
@@ -244,7 +244,7 @@ class TestOutput:
|
||||
factory = WorkflowFactory(mcp_tool_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(output={"result": "Local.Result"})))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["Result"] == [{"k": "v", "n": 1}]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -253,7 +253,7 @@ class TestOutput:
|
||||
factory = WorkflowFactory(mcp_tool_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(output={"result": "Local.Result"})))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["Result"] == ["plain text not json"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -262,7 +262,7 @@ class TestOutput:
|
||||
factory = WorkflowFactory(mcp_tool_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(output={"messages": "Local.Messages"})))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
msg = decl["Local"]["Messages"]
|
||||
# Single Tool-role message containing both contents (parity with .NET).
|
||||
assert isinstance(msg, Message)
|
||||
@@ -276,7 +276,7 @@ class TestOutput:
|
||||
factory = WorkflowFactory(mcp_tool_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(output={"result": "Local.Result"})))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["Result"] == ["https://example.com/file.txt"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -285,7 +285,7 @@ class TestOutput:
|
||||
factory = WorkflowFactory(mcp_tool_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(output={"result": {"path": "Local.Result"}})))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["Result"] == ["ok"]
|
||||
|
||||
|
||||
@@ -306,7 +306,7 @@ class TestConversation:
|
||||
)
|
||||
)
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
conv = decl["System"]["conversations"]["conv-42"]
|
||||
msgs = conv["messages"] if isinstance(conv, dict) else conv.messages
|
||||
assert len(msgs) == 1
|
||||
@@ -328,7 +328,7 @@ class TestConversation:
|
||||
)
|
||||
)
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
# Empty conversation id must not produce a `""` entry under System.conversations.
|
||||
conversations = decl.get("System", {}).get("conversations", {})
|
||||
assert "" not in conversations
|
||||
@@ -529,7 +529,7 @@ class TestErrorHandling:
|
||||
factory = WorkflowFactory(mcp_tool_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(output={"result": "Local.Result"})))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["Result"] == "Error: server down"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -538,7 +538,7 @@ class TestErrorHandling:
|
||||
factory = WorkflowFactory(mcp_tool_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(output={"result": "Local.Result"})))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert decl["Local"]["Result"] == "Error: invalid arguments"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -547,7 +547,7 @@ class TestErrorHandling:
|
||||
factory = WorkflowFactory(mcp_tool_handler=handler)
|
||||
workflow = factory.create_workflow_from_definition(_yaml(_action(output={"result": "Local.Result"})))
|
||||
await workflow.run({})
|
||||
decl = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
decl = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
result = decl["Local"]["Result"]
|
||||
assert isinstance(result, str)
|
||||
assert result.startswith("Error:")
|
||||
|
||||
@@ -291,11 +291,11 @@ actions:
|
||||
# Stamp a marker into the declarative state between turns. The
|
||||
# continuation branch must preserve it; a state-clearing run would
|
||||
# wipe ``DECLARATIVE_STATE_KEY`` and force re-initialization.
|
||||
state_data = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
state_data = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert isinstance(state_data, dict), "Expected declarative state to be initialized after turn 1"
|
||||
state_data["Local"] = {"persisted_marker": "kept-from-turn-1"}
|
||||
workflow._runner.state.set(DECLARATIVE_STATE_KEY, state_data)
|
||||
workflow._runner.state.commit()
|
||||
workflow._state.set(DECLARATIVE_STATE_KEY, state_data)
|
||||
workflow._state.commit()
|
||||
|
||||
second = await agent.run("turn-2-msg")
|
||||
assert second.text == "turn-2-msg", (
|
||||
@@ -305,7 +305,7 @@ actions:
|
||||
# The continuation branch in ``_ensure_state_initialized`` must:
|
||||
# 1. preserve the cross-turn marker we stamped above
|
||||
# 2. refresh Inputs.input and System.LastMessage* to the new turn
|
||||
post_state = workflow._runner.state.get(DECLARATIVE_STATE_KEY)
|
||||
post_state = workflow._state.get(DECLARATIVE_STATE_KEY)
|
||||
assert isinstance(post_state, dict), "declarative state vanished between turns"
|
||||
local = post_state.get("Local", {})
|
||||
assert local.get("persisted_marker") == "kept-from-turn-1", (
|
||||
|
||||
@@ -1,21 +0,0 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) Microsoft Corporation.
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE
|
||||
@@ -1,21 +0,0 @@
|
||||
# agent-framework-hosting-responses
|
||||
|
||||
OpenAI Responses-shaped channel for `agent-framework-hosting`.
|
||||
|
||||
Exposes a single `POST /responses` endpoint that accepts the OpenAI
|
||||
Responses API request body and returns either a Responses-shaped JSON
|
||||
body or a Server-Sent-Events stream when `stream=True`.
|
||||
|
||||
```python
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
from agent_framework_hosting import AgentFrameworkHost
|
||||
from agent_framework_hosting_responses import ResponsesChannel
|
||||
|
||||
agent = OpenAIChatClient().as_agent(name="Assistant")
|
||||
|
||||
host = AgentFrameworkHost(target=agent, channels=[ResponsesChannel()])
|
||||
host.serve(port=8000)
|
||||
```
|
||||
|
||||
The base host plumbing lives in
|
||||
[`agent-framework-hosting`](https://pypi.org/project/agent-framework-hosting/).
|
||||
@@ -1,25 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""OpenAI Responses-shaped channel for ``agent-framework-hosting``."""
|
||||
|
||||
import importlib.metadata
|
||||
|
||||
from ._channel import ResponsesChannel
|
||||
from ._parsing import (
|
||||
messages_from_responses_input,
|
||||
parse_responses_identity,
|
||||
parse_responses_request,
|
||||
)
|
||||
|
||||
try:
|
||||
__version__ = importlib.metadata.version(__name__)
|
||||
except importlib.metadata.PackageNotFoundError:
|
||||
__version__ = "0.0.0"
|
||||
|
||||
__all__ = [
|
||||
"ResponsesChannel",
|
||||
"__version__",
|
||||
"messages_from_responses_input",
|
||||
"parse_responses_identity",
|
||||
"parse_responses_request",
|
||||
]
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,156 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Parsing helpers for the OpenAI Responses-API request body.
|
||||
|
||||
The Responses API accepts ``input`` as either a string or a list of "input
|
||||
items". An item is either a content part (``input_text`` / ``input_image``
|
||||
/ ``input_file``) or a message envelope ``{type: "message", role,
|
||||
content: [...]}``. We translate that into an Agent Framework ``Message``
|
||||
list and remap the generation-control fields the API also carries into
|
||||
``ChatOptions``-shaped keys. The result is available to the channel's
|
||||
``run_hook``; a default hook strips them before they reach the agent so
|
||||
unknown fields from untrusted callers are not forwarded unless the host
|
||||
developer explicitly opts in.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, cast
|
||||
|
||||
from agent_framework import Content, Message
|
||||
from agent_framework_hosting import ChannelIdentity, ChannelSession
|
||||
|
||||
# OpenAI Responses field name → Agent Framework ChatOptions field name.
|
||||
_RESPONSES_OPTION_REMAP = {
|
||||
"max_output_tokens": "max_tokens",
|
||||
"parallel_tool_calls": "allow_multiple_tool_calls",
|
||||
}
|
||||
# Fields the Responses transport owns; they are consumed separately and must
|
||||
# not also appear in options.
|
||||
_RESPONSES_TRANSPORT_KEYS = frozenset({"input", "stream", "previous_response_id"})
|
||||
|
||||
|
||||
def parse_responses_identity(body: Mapping[str, Any], channel_name: str) -> ChannelIdentity | None:
|
||||
"""Surface the caller as a :class:`ChannelIdentity` so the host can record it.
|
||||
|
||||
OpenAI Responses replaced ``user`` with ``safety_identifier`` — we use
|
||||
that as the native id, falling back to the legacy ``user`` field.
|
||||
"""
|
||||
native = body.get("safety_identifier") or body.get("user")
|
||||
if not isinstance(native, str) or not native:
|
||||
return None
|
||||
return ChannelIdentity(channel=channel_name, native_id=native)
|
||||
|
||||
|
||||
def _content_from_input_item(item: Mapping[str, Any]) -> Content:
|
||||
"""Convert a single OpenAI Responses ``input`` item into a :class:`Content` part.
|
||||
|
||||
Handles the ``input_text``/``output_text``/``text`` text variants,
|
||||
``input_image`` URL references, and ``input_file`` references via either
|
||||
a public URL or a hosted ``file_id``. Raises ``ValueError`` for any
|
||||
unsupported item type so the surrounding parser can return a 422.
|
||||
"""
|
||||
item_type = item.get("type")
|
||||
if item_type in ("input_text", "output_text", "text"):
|
||||
return Content.from_text(text=str(item.get("text", "")))
|
||||
if item_type == "input_image":
|
||||
image_url: Any = item.get("image_url")
|
||||
if isinstance(image_url, Mapping):
|
||||
image_url = cast("Mapping[str, Any]", image_url).get("url")
|
||||
if not isinstance(image_url, str):
|
||||
raise ValueError("input_image requires `image_url`")
|
||||
return Content.from_uri(uri=image_url, media_type="image/*")
|
||||
if item_type == "input_file":
|
||||
if (uri := item.get("file_url")) and isinstance(uri, str):
|
||||
return Content.from_uri(uri=uri, media_type=item.get("mime_type"))
|
||||
if file_id := item.get("file_id"):
|
||||
return Content(type="hosted_file", file_id=str(file_id))
|
||||
raise ValueError("input_file requires `file_url` or `file_id`")
|
||||
raise ValueError(f"Unsupported Responses input content type: {item_type!r}")
|
||||
|
||||
|
||||
def messages_from_responses_input(value: Any) -> list[Message]:
|
||||
"""Translate ``input`` (string or list of items) into :class:`Message` objects."""
|
||||
if isinstance(value, str):
|
||||
return [Message("user", [Content.from_text(text=value)])]
|
||||
if not isinstance(value, list) or not value:
|
||||
raise ValueError("`input` must be a non-empty string or list")
|
||||
|
||||
messages: list[Message] = []
|
||||
pending_user_parts: list[Content] = []
|
||||
|
||||
def flush() -> None:
|
||||
"""Emit any buffered loose user content as a single user message."""
|
||||
if pending_user_parts:
|
||||
messages.append(Message("user", list(pending_user_parts)))
|
||||
pending_user_parts.clear()
|
||||
|
||||
for item in cast("list[Any]", value):
|
||||
if not isinstance(item, Mapping):
|
||||
raise ValueError("each `input` item must be an object")
|
||||
item_map = cast("Mapping[str, Any]", item)
|
||||
if item_map.get("type") == "message":
|
||||
flush()
|
||||
role = str(item_map.get("role") or "user")
|
||||
content: Any = item_map.get("content") or []
|
||||
parts: list[Content]
|
||||
if isinstance(content, str):
|
||||
parts = [Content.from_text(text=content)]
|
||||
elif isinstance(content, list):
|
||||
parts = []
|
||||
for content_item in cast("list[Any]", content):
|
||||
if not isinstance(content_item, Mapping):
|
||||
raise ValueError("each message `content` item must be an object")
|
||||
parts.append(_content_from_input_item(cast("Mapping[str, Any]", content_item)))
|
||||
else:
|
||||
raise ValueError("message `content` must be a string or list")
|
||||
messages.append(Message(role, parts))
|
||||
else:
|
||||
pending_user_parts.append(_content_from_input_item(item_map))
|
||||
|
||||
flush()
|
||||
if not messages:
|
||||
raise ValueError("`input` produced no messages")
|
||||
return messages
|
||||
|
||||
|
||||
def parse_responses_request(
|
||||
body: Mapping[str, Any],
|
||||
) -> tuple[list[Message], dict[str, Any], ChannelSession | None]:
|
||||
"""Translate a Responses-API request body into Agent Framework constructs.
|
||||
|
||||
Returns a triple ``(messages, options, session)`` where:
|
||||
|
||||
- ``messages`` is the parsed conversation.
|
||||
- ``options`` is a ``ChatOptions``-shaped dict with the remapped
|
||||
generation-control fields. Known Responses→ChatOptions renames are
|
||||
applied (e.g. ``max_output_tokens`` → ``max_tokens``); transport/
|
||||
session keys are excluded; ``None``-valued fields are dropped.
|
||||
Unknown fields are forwarded as-is so the channel's ``run_hook``
|
||||
can inspect and filter them. The default ``ResponsesChannel`` strips
|
||||
all options before the agent runs; supply a custom ``run_hook`` to
|
||||
selectively keep fields.
|
||||
- ``session`` is a :class:`ChannelSession` keyed by
|
||||
``previous_response_id`` when one was supplied, else ``None``.
|
||||
"""
|
||||
messages = messages_from_responses_input(body.get("input"))
|
||||
|
||||
options: dict[str, Any] = {}
|
||||
for key, value in body.items():
|
||||
if key in _RESPONSES_TRANSPORT_KEYS or value is None:
|
||||
continue
|
||||
options[_RESPONSES_OPTION_REMAP.get(key, key)] = value
|
||||
|
||||
session: ChannelSession | None = None
|
||||
if (prev := body.get("previous_response_id")) and isinstance(prev, str):
|
||||
session = ChannelSession(isolation_key=prev)
|
||||
|
||||
return messages, options, session
|
||||
|
||||
|
||||
__all__ = [
|
||||
"messages_from_responses_input",
|
||||
"parse_responses_identity",
|
||||
"parse_responses_request",
|
||||
]
|
||||
@@ -1,80 +0,0 @@
|
||||
[project]
|
||||
name = "agent-framework-hosting-responses"
|
||||
description = "OpenAI Responses-shaped channel for agent-framework-hosting."
|
||||
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.10"
|
||||
version = "1.0.0a260424"
|
||||
license-files = ["LICENSE"]
|
||||
urls.homepage = "https://aka.ms/agent-framework"
|
||||
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
|
||||
urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true"
|
||||
urls.issues = "https://github.com/microsoft/agent-framework/issues"
|
||||
classifiers = [
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Development Status :: 3 - Alpha",
|
||||
"Intended Audience :: Developers",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Programming Language :: Python :: 3.13",
|
||||
"Programming Language :: Python :: 3.14",
|
||||
"Typing :: Typed",
|
||||
]
|
||||
dependencies = [
|
||||
"agent-framework-core>=1.2.0,<2",
|
||||
"agent-framework-hosting==1.0.0a260424",
|
||||
"openai>=1.99.0,<3",
|
||||
]
|
||||
|
||||
[tool.uv]
|
||||
prerelease = "if-necessary-or-explicit"
|
||||
environments = [
|
||||
"sys_platform == 'darwin'",
|
||||
"sys_platform == 'linux'",
|
||||
"sys_platform == 'win32'"
|
||||
]
|
||||
|
||||
[tool.uv-dynamic-versioning]
|
||||
fallback-version = "0.0.0"
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = 'tests'
|
||||
addopts = "-ra -q -r fEX"
|
||||
asyncio_mode = "auto"
|
||||
asyncio_default_fixture_loop_scope = "function"
|
||||
filterwarnings = []
|
||||
timeout = 120
|
||||
markers = [
|
||||
"integration: marks tests as integration tests that require external services",
|
||||
]
|
||||
|
||||
[tool.ruff]
|
||||
extend = "../../pyproject.toml"
|
||||
|
||||
[tool.coverage.run]
|
||||
omit = [
|
||||
"**/__init__.py"
|
||||
]
|
||||
|
||||
[tool.pyright]
|
||||
extends = "../../pyproject.toml"
|
||||
include = ["agent_framework_hosting_responses"]
|
||||
exclude = ['tests']
|
||||
|
||||
[tool.bandit]
|
||||
targets = ["agent_framework_hosting_responses"]
|
||||
exclude_dirs = ["tests"]
|
||||
|
||||
[tool.poe]
|
||||
executor.type = "uv"
|
||||
include = "../../shared_tasks.toml"
|
||||
|
||||
[tool.poe.tasks.test]
|
||||
help = "Run the default unit test suite for this package."
|
||||
cmd = 'pytest -m "not integration" --cov=agent_framework_hosting_responses --cov-report=term-missing:skip-covered tests'
|
||||
|
||||
[build-system]
|
||||
requires = ["flit-core >= 3.11,<4.0"]
|
||||
build-backend = "flit_core.buildapi"
|
||||
@@ -1,651 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""End-to-end tests for :class:`ResponsesChannel` via Starlette's ``TestClient``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from collections.abc import AsyncIterator
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from agent_framework import AgentResponse, AgentResponseUpdate, Content, Message
|
||||
from agent_framework_hosting import (
|
||||
AgentFrameworkHost,
|
||||
HostedRunResult,
|
||||
)
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from agent_framework_hosting_responses import ResponsesChannel
|
||||
from agent_framework_hosting_responses._channel import ( # pyright: ignore[reportPrivateUsage]
|
||||
_result_to_output_items,
|
||||
_result_to_text,
|
||||
)
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Fakes #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeAgentResponse:
|
||||
text: str
|
||||
|
||||
|
||||
class _FakeStream:
|
||||
"""Minimal stand-in for AF's ``ResponseStream`` returned by ``run(stream=True)``."""
|
||||
|
||||
def __init__(self, chunks: list[str]) -> None:
|
||||
self._chunks = chunks
|
||||
self._final = _FakeAgentResponse(text="".join(chunks))
|
||||
|
||||
def __aiter__(self) -> AsyncIterator[AgentResponseUpdate]:
|
||||
async def _gen() -> AsyncIterator[AgentResponseUpdate]:
|
||||
for c in self._chunks:
|
||||
yield AgentResponseUpdate(contents=[Content.from_text(c)], role="assistant")
|
||||
|
||||
return _gen()
|
||||
|
||||
async def get_final_response(self) -> _FakeAgentResponse:
|
||||
return self._final
|
||||
|
||||
|
||||
class _FakeAgent:
|
||||
def __init__(self, reply: Any = "hello", chunks: list[str] | None = None) -> None:
|
||||
self.id = "fake-agent"
|
||||
self.name: str | None = "Fake Agent"
|
||||
self.description: str | None = "Test fake agent"
|
||||
self._reply = reply
|
||||
self._chunks = chunks or [reply]
|
||||
self.calls: list[dict[str, Any]] = []
|
||||
|
||||
def create_session(self, *, session_id: str | None = None) -> Any:
|
||||
return {"session_id": session_id}
|
||||
|
||||
def get_session(self, service_session_id: str, *, session_id: str | None = None) -> Any:
|
||||
return {"service_session_id": service_session_id, "session_id": session_id}
|
||||
|
||||
def run(self, messages: Any = None, *, stream: bool = False, **kwargs: Any) -> Any:
|
||||
self.calls.append({"messages": messages, "stream": stream, "kwargs": kwargs})
|
||||
if stream:
|
||||
return _FakeStream(self._chunks)
|
||||
|
||||
async def _coro() -> Any:
|
||||
if not isinstance(self._reply, str):
|
||||
return self._reply
|
||||
return _FakeAgentResponse(text=self._reply)
|
||||
|
||||
return _coro()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Tests #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _make_client(
|
||||
agent: _FakeAgent | None = None,
|
||||
*,
|
||||
path: str = "/responses",
|
||||
response_id_factory: Any | None = None,
|
||||
) -> tuple[TestClient, AgentFrameworkHost, _FakeAgent]:
|
||||
agent = agent or _FakeAgent()
|
||||
host = AgentFrameworkHost(
|
||||
target=agent,
|
||||
channels=[ResponsesChannel(path=path, response_id_factory=response_id_factory)],
|
||||
)
|
||||
return TestClient(host.app), host, agent
|
||||
|
||||
|
||||
def _sse_payload(body: str, event_type: str) -> dict[str, Any]:
|
||||
current_event: str | None = None
|
||||
for line in body.splitlines():
|
||||
if line.startswith("event: "):
|
||||
current_event = line[len("event: ") :]
|
||||
continue
|
||||
if current_event == event_type and line.startswith("data: "):
|
||||
return json.loads(line[len("data: ") :])
|
||||
raise AssertionError(f"Missing SSE event: {event_type}")
|
||||
|
||||
|
||||
class TestResponsesChannelNonStreaming:
|
||||
def test_post_responses_returns_completed_envelope(self) -> None:
|
||||
client, _host, agent = _make_client(_FakeAgent(reply="hi back"))
|
||||
with client:
|
||||
r = client.post("/responses", json={"input": "hi"})
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["status"] == "completed"
|
||||
assert body["object"] == "response"
|
||||
assert body["id"].startswith("resp_")
|
||||
assert isinstance(body["created_at"], int)
|
||||
assert body["output"][0]["content"][0]["text"] == "hi back"
|
||||
assert len(agent.calls) == 1
|
||||
|
||||
def test_non_string_model_falls_back_to_agent(self) -> None:
|
||||
client, _host, _agent = _make_client(_FakeAgent(reply="hi"))
|
||||
with client:
|
||||
r = client.post("/responses", json={"input": "hi", "model": None})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["model"] == "agent"
|
||||
|
||||
def test_empty_path_mounts_at_app_root(self) -> None:
|
||||
client, _host, _agent = _make_client(_FakeAgent(reply="hi back"), path="")
|
||||
with client:
|
||||
r = client.post("/", json={"input": "hi"})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["output"][0]["content"][0]["text"] == "hi back"
|
||||
|
||||
def test_custom_path_mounts_route_under_host_path(self) -> None:
|
||||
client, _host, _agent = _make_client(_FakeAgent(reply="custom"), path="/api/responses")
|
||||
with client:
|
||||
r = client.post("/api/responses", json={"input": "hi"})
|
||||
missing = client.post("/api/responses/responses", json={"input": "hi"})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["output"][0]["content"][0]["text"] == "custom"
|
||||
assert missing.status_code == 404
|
||||
|
||||
def test_invalid_json_returns_400(self) -> None:
|
||||
client, *_ = _make_client()
|
||||
with client:
|
||||
r = client.post("/responses", content=b"{not json", headers={"content-type": "application/json"})
|
||||
assert r.status_code == 400
|
||||
|
||||
def test_non_object_json_returns_422(self) -> None:
|
||||
client, *_ = _make_client()
|
||||
with client:
|
||||
r = client.post("/responses", json=["not", "an", "object"])
|
||||
assert r.status_code == 422
|
||||
assert r.json()["error"] == "request body must be a JSON object"
|
||||
|
||||
def test_invalid_input_returns_422(self) -> None:
|
||||
client, *_ = _make_client()
|
||||
with client:
|
||||
r = client.post("/responses", json={"input": 42})
|
||||
assert r.status_code == 422
|
||||
|
||||
def test_request_options_are_not_forwarded_by_default(self) -> None:
|
||||
client, _host, agent = _make_client()
|
||||
with client:
|
||||
r = client.post(
|
||||
"/responses",
|
||||
json={"input": "x", "temperature": 0.5, "max_output_tokens": 64, "truncation": "auto"},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
assert "options" not in agent.calls[0]["kwargs"]
|
||||
|
||||
def test_custom_run_hook_can_forward_options(self) -> None:
|
||||
import dataclasses
|
||||
|
||||
def keep_temperature(request: Any, **_: Any) -> Any:
|
||||
opts = dict(request.options or {})
|
||||
return dataclasses.replace(request, options={"temperature": opts.get("temperature")})
|
||||
|
||||
agent = _FakeAgent()
|
||||
host = AgentFrameworkHost(
|
||||
target=agent,
|
||||
channels=[ResponsesChannel(run_hook=keep_temperature)],
|
||||
)
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/responses", json={"input": "x", "temperature": 0.7, "truncation": "auto"})
|
||||
assert r.status_code == 200
|
||||
opts = agent.calls[0]["kwargs"]["options"]
|
||||
assert opts == {"temperature": 0.7}
|
||||
assert "truncation" not in opts
|
||||
|
||||
def test_multimodal_agent_response_outputs_are_preserved(self) -> None:
|
||||
response = AgentResponse(
|
||||
messages=[
|
||||
Message(
|
||||
"assistant",
|
||||
[
|
||||
Content.from_text_reasoning(id="rs_1", text="checking"),
|
||||
Content.from_function_call("call_1", "collect_media", arguments={"city": "Seattle"}),
|
||||
Content.from_function_result(
|
||||
"call_1",
|
||||
result=[
|
||||
Content.from_text("caption"),
|
||||
Content.from_uri("https://example.com/cat.png", media_type="image/png"),
|
||||
Content.from_hosted_file("file_pdf", media_type="application/pdf"),
|
||||
],
|
||||
),
|
||||
Content.from_text("done"),
|
||||
],
|
||||
)
|
||||
],
|
||||
)
|
||||
client, _host, _agent = _make_client(_FakeAgent(reply=response))
|
||||
|
||||
with client:
|
||||
r = client.post("/responses", json={"input": "hi"})
|
||||
|
||||
assert r.status_code == 200
|
||||
output = r.json()["output"]
|
||||
assert [item["type"] for item in output] == [
|
||||
"reasoning",
|
||||
"function_call",
|
||||
"function_call_output",
|
||||
"message",
|
||||
]
|
||||
assert output[0]["content"][0]["text"] == "checking"
|
||||
assert output[1]["name"] == "collect_media"
|
||||
assert output[1]["arguments"] == '{"city": "Seattle"}'
|
||||
assert output[2]["output"] == [
|
||||
{"text": "caption", "type": "input_text"},
|
||||
{"detail": "auto", "type": "input_image", "image_url": "https://example.com/cat.png"},
|
||||
{"type": "input_file", "file_id": "file_pdf"},
|
||||
]
|
||||
assert output[3]["content"][0]["text"] == "done"
|
||||
|
||||
def test_raw_responses_output_items_are_preserved(self) -> None:
|
||||
raw_item = {
|
||||
"id": "ig_1",
|
||||
"type": "image_generation_call",
|
||||
"result": "base64-image",
|
||||
"status": "completed",
|
||||
}
|
||||
response = AgentResponse(
|
||||
messages=[
|
||||
Message(
|
||||
"assistant",
|
||||
[
|
||||
Content.from_image_generation_tool_call(image_id="ig_1", raw_representation=raw_item),
|
||||
Content.from_image_generation_tool_result(
|
||||
image_id="ig_1",
|
||||
outputs=Content.from_uri("data:image/png;base64,base64-image", media_type="image/png"),
|
||||
raw_representation=raw_item,
|
||||
),
|
||||
],
|
||||
)
|
||||
],
|
||||
)
|
||||
client, _host, _agent = _make_client(_FakeAgent(reply=response))
|
||||
|
||||
with client:
|
||||
r = client.post("/responses", json={"input": "hi"})
|
||||
|
||||
assert r.status_code == 200
|
||||
assert r.json()["output"] == [raw_item]
|
||||
|
||||
def test_later_raw_responses_output_item_replaces_earlier_partial_item(self) -> None:
|
||||
partial = {
|
||||
"id": "mcp_1",
|
||||
"type": "mcp_call",
|
||||
"server_label": "weather",
|
||||
"name": "lookup",
|
||||
"arguments": "{}",
|
||||
"status": "in_progress",
|
||||
}
|
||||
completed = {**partial, "status": "completed", "output": "sunny"}
|
||||
response = AgentResponse(
|
||||
messages=[
|
||||
Message(
|
||||
"assistant",
|
||||
[
|
||||
Content.from_mcp_server_tool_call(
|
||||
"mcp_1",
|
||||
"lookup",
|
||||
server_name="weather",
|
||||
raw_representation=partial,
|
||||
),
|
||||
Content.from_mcp_server_tool_result("mcp_1", output="sunny", raw_representation=completed),
|
||||
],
|
||||
)
|
||||
],
|
||||
)
|
||||
client, _host, _agent = _make_client(_FakeAgent(reply=response))
|
||||
|
||||
with client:
|
||||
r = client.post("/responses", json={"input": "hi"})
|
||||
|
||||
assert r.status_code == 200
|
||||
assert r.json()["output"] == [completed]
|
||||
|
||||
def test_previous_response_id_creates_session(self) -> None:
|
||||
client, _host, agent = _make_client()
|
||||
with client:
|
||||
client.post("/responses", json={"input": "x", "previous_response_id": "resp_42"})
|
||||
# AgentFrameworkHost converts the channel session into an AgentSession.
|
||||
sess = agent.calls[0]["kwargs"].get("session")
|
||||
assert sess is not None
|
||||
# _FakeAgent.create_session stashes the session_id on the dict it returns.
|
||||
assert sess["session_id"] == "resp_42"
|
||||
|
||||
def test_first_turn_response_id_creates_session(self) -> None:
|
||||
client, _host, agent = _make_client(response_id_factory=lambda *_: "resp_first")
|
||||
with client:
|
||||
client.post("/responses", json={"input": "x"})
|
||||
sess = agent.calls[0]["kwargs"].get("session")
|
||||
assert sess is not None
|
||||
assert sess["session_id"] == "resp_first"
|
||||
|
||||
def test_chat_isolation_header_ignored_outside_foundry(self) -> None:
|
||||
client, _host, agent = _make_client(response_id_factory=lambda *_: "resp_local")
|
||||
with client:
|
||||
client.post(
|
||||
"/responses",
|
||||
json={"input": "x"},
|
||||
headers={"x-agent-chat-isolation-key": "chat-abc"},
|
||||
)
|
||||
sess = agent.calls[0]["kwargs"].get("session")
|
||||
assert sess is not None
|
||||
assert sess["session_id"] == "resp_local"
|
||||
|
||||
def test_chat_isolation_header_creates_session_in_foundry(self, monkeypatch: Any) -> None:
|
||||
"""Foundry-style ``x-agent-chat-isolation-key`` falls back to a session anchor.
|
||||
|
||||
First-turn requests have no ``previous_response_id`` (the client
|
||||
doesn't have one yet), but Foundry Hosted Agents always inject
|
||||
the isolation headers. The channel must derive a session from the
|
||||
chat key so the host can build a stable per-conversation session
|
||||
that history providers persist under.
|
||||
"""
|
||||
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
|
||||
client, _host, agent = _make_client()
|
||||
with client:
|
||||
client.post(
|
||||
"/responses",
|
||||
json={"input": "x"},
|
||||
headers={"x-agent-chat-isolation-key": "chat-abc"},
|
||||
)
|
||||
sess = agent.calls[0]["kwargs"].get("session")
|
||||
assert sess is not None
|
||||
assert sess["session_id"] == "chat-abc"
|
||||
|
||||
def test_prev_response_id_wins_over_chat_isolation_header(self, monkeypatch: Any) -> None:
|
||||
"""When both anchors are present, ``previous_response_id`` wins.
|
||||
|
||||
``previous_response_id`` is the protocol-native chain anchor; the
|
||||
header fallback is only meant to bootstrap when no protocol
|
||||
anchor exists.
|
||||
"""
|
||||
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
|
||||
client, _host, agent = _make_client()
|
||||
with client:
|
||||
client.post(
|
||||
"/responses",
|
||||
json={"input": "x", "previous_response_id": "resp_99"},
|
||||
headers={"x-agent-chat-isolation-key": "chat-abc"},
|
||||
)
|
||||
sess = agent.calls[0]["kwargs"].get("session")
|
||||
assert sess is not None
|
||||
assert sess["session_id"] == "resp_99"
|
||||
|
||||
def test_response_hook_can_rewrite_originating_reply(self) -> None:
|
||||
seen_kwargs: list[dict[str, Any]] = []
|
||||
|
||||
def hook(result: HostedRunResult, **kwargs: Any) -> HostedRunResult:
|
||||
seen_kwargs.append(dict(kwargs))
|
||||
return HostedRunResult(_FakeAgentResponse(text=result.result.text.upper()), session=result.session)
|
||||
|
||||
agent = _FakeAgent(reply="hooked")
|
||||
host = AgentFrameworkHost(target=agent, channels=[ResponsesChannel(response_hook=hook)])
|
||||
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/responses", json={"input": "hi"})
|
||||
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["output"][0]["content"][0]["text"] == "HOOKED"
|
||||
assert seen_kwargs
|
||||
assert seen_kwargs[0]["channel_name"] == "responses"
|
||||
|
||||
|
||||
class TestResultTextRendering:
|
||||
def test_result_text_prefers_text_property(self) -> None:
|
||||
assert _result_to_text(_FakeAgentResponse(text="plain")) == "plain"
|
||||
|
||||
def test_result_text_projects_workflow_outputs(self) -> None:
|
||||
class _WorkflowResult:
|
||||
def get_outputs(self) -> list[Any]:
|
||||
return [_FakeAgentResponse(text="one"), " two"]
|
||||
|
||||
assert _result_to_text(_WorkflowResult()) == "one two"
|
||||
|
||||
def test_result_output_items_project_workflow_message_and_content_outputs(self) -> None:
|
||||
class _WorkflowResult:
|
||||
def get_outputs(self) -> list[Any]:
|
||||
return [
|
||||
Message("assistant", [Content.from_text("one")]),
|
||||
Content.from_function_result(
|
||||
"call_1",
|
||||
result=[Content.from_uri("https://example.com/cat.png", media_type="image/png")],
|
||||
),
|
||||
]
|
||||
|
||||
output = [
|
||||
item.model_dump(mode="json", exclude_none=True)
|
||||
for item in _result_to_output_items(_WorkflowResult(), status="completed")
|
||||
]
|
||||
assert output[0]["type"] == "message"
|
||||
assert output[0]["content"][0]["text"] == "one"
|
||||
assert output[1]["type"] == "function_call_output"
|
||||
assert output[1]["output"] == [
|
||||
{"detail": "auto", "type": "input_image", "image_url": "https://example.com/cat.png"}
|
||||
]
|
||||
|
||||
def test_function_result_exception_is_preserved(self) -> None:
|
||||
output = [
|
||||
item.model_dump(mode="json", exclude_none=True)
|
||||
for item in _result_to_output_items(
|
||||
Content.from_function_result("call_1", exception="tool failed"),
|
||||
status="completed",
|
||||
)
|
||||
]
|
||||
assert output[0]["output"] == "tool failed"
|
||||
|
||||
def test_stateful_call_and_result_content_coalesce_to_one_output_item(self) -> None:
|
||||
output = [
|
||||
item.model_dump(mode="json", exclude_none=True)
|
||||
for item in _result_to_output_items(
|
||||
Message(
|
||||
"assistant",
|
||||
[
|
||||
Content.from_image_generation_tool_call(image_id="ig_1"),
|
||||
Content.from_image_generation_tool_result(
|
||||
image_id="ig_1",
|
||||
outputs=Content.from_uri("data:image/png;base64,base64-image", media_type="image/png"),
|
||||
),
|
||||
Content.from_mcp_server_tool_call(
|
||||
"mcp_1",
|
||||
"lookup",
|
||||
server_name="weather",
|
||||
arguments={"city": "Seattle"},
|
||||
),
|
||||
Content.from_mcp_server_tool_result("mcp_1", output=[Content.from_text("sunny")]),
|
||||
],
|
||||
),
|
||||
status="completed",
|
||||
)
|
||||
]
|
||||
assert output == [
|
||||
{
|
||||
"id": "ig_1",
|
||||
"result": "base64-image",
|
||||
"status": "completed",
|
||||
"type": "image_generation_call",
|
||||
},
|
||||
{
|
||||
"id": "mcp_1",
|
||||
"arguments": '{"city": "Seattle"}',
|
||||
"name": "lookup",
|
||||
"output": "sunny",
|
||||
"server_label": "weather",
|
||||
"status": "completed",
|
||||
"type": "mcp_call",
|
||||
},
|
||||
]
|
||||
|
||||
def test_stateful_call_and_result_content_coalesce_across_messages(self) -> None:
|
||||
output = [
|
||||
item.model_dump(mode="json", exclude_none=True)
|
||||
for item in _result_to_output_items(
|
||||
AgentResponse(
|
||||
messages=[
|
||||
Message(
|
||||
"assistant",
|
||||
[
|
||||
Content.from_mcp_server_tool_call(
|
||||
"mcp_1",
|
||||
"lookup",
|
||||
server_name="weather",
|
||||
arguments={"city": "Seattle"},
|
||||
)
|
||||
],
|
||||
),
|
||||
Message(
|
||||
"tool",
|
||||
[Content.from_mcp_server_tool_result("mcp_1", output=[Content.from_text("sunny")])],
|
||||
),
|
||||
]
|
||||
),
|
||||
status="completed",
|
||||
)
|
||||
]
|
||||
assert output == [
|
||||
{
|
||||
"id": "mcp_1",
|
||||
"arguments": '{"city": "Seattle"}',
|
||||
"name": "lookup",
|
||||
"output": "sunny",
|
||||
"server_label": "weather",
|
||||
"status": "completed",
|
||||
"type": "mcp_call",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
class TestResponsesChannelStreaming:
|
||||
def test_sse_emits_created_delta_completed(self) -> None:
|
||||
agent = _FakeAgent(reply="hello world", chunks=["hello", " ", "world"])
|
||||
host = AgentFrameworkHost(target=agent, channels=[ResponsesChannel()])
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/responses", json={"input": "hi", "stream": True})
|
||||
assert r.status_code == 200
|
||||
body = r.text
|
||||
|
||||
# SSE event lines look like "event: <type>\ndata: <json>\n\n".
|
||||
events = [line[len("event: ") :] for line in body.splitlines() if line.startswith("event: ")]
|
||||
assert events[0] == "response.created"
|
||||
assert events[-1] == "response.completed"
|
||||
assert events.count("response.output_text.delta") == 3
|
||||
|
||||
def test_sse_transform_hook_can_rewrite_chunks(self) -> None:
|
||||
agent = _FakeAgent(reply="hello", chunks=["he", "llo"])
|
||||
|
||||
def transform(update: AgentResponseUpdate) -> AgentResponseUpdate:
|
||||
return AgentResponseUpdate(contents=[Content.from_text(update.text.upper())], role="assistant")
|
||||
|
||||
host = AgentFrameworkHost(target=agent, channels=[ResponsesChannel(stream_update_hook=transform)])
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/responses", json={"input": "hi", "stream": True})
|
||||
|
||||
assert r.status_code == 200
|
||||
assert '"delta":"HE"' in r.text
|
||||
assert '"delta":"LLO"' in r.text
|
||||
# Stream update hooks are update-only; they do not rewrite get_final_response().
|
||||
assert '"text":"hello"' in r.text
|
||||
|
||||
def test_sse_completed_preserves_streamed_multimodal_updates_when_finalize_fails(self) -> None:
|
||||
class _MultimodalStream:
|
||||
def __aiter__(self) -> AsyncIterator[AgentResponseUpdate]:
|
||||
async def _gen() -> AsyncIterator[AgentResponseUpdate]:
|
||||
yield AgentResponseUpdate(
|
||||
contents=[
|
||||
Content.from_text("caption"),
|
||||
Content.from_text_reasoning(id="rs_1", text="thinking"),
|
||||
Content.from_function_call("call_1", "lookup", arguments={"city": "Seattle"}),
|
||||
Content.from_uri("https://example.com/cat.png", media_type="image/png"),
|
||||
],
|
||||
role="assistant",
|
||||
)
|
||||
|
||||
return _gen()
|
||||
|
||||
async def get_final_response(self) -> _FakeAgentResponse:
|
||||
raise RuntimeError("finalize unavailable")
|
||||
|
||||
class _MultimodalAgent(_FakeAgent):
|
||||
def run(self, messages: Any = None, *, stream: bool = False, **kwargs: Any) -> Any:
|
||||
self.calls.append({"messages": messages, "stream": stream, "kwargs": kwargs})
|
||||
if stream:
|
||||
return _MultimodalStream()
|
||||
raise AssertionError("non-streaming path not exercised here")
|
||||
|
||||
host = AgentFrameworkHost(target=_MultimodalAgent(), channels=[ResponsesChannel()])
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/responses", json={"input": "hi", "stream": True})
|
||||
|
||||
assert r.status_code == 200
|
||||
assert "event: response.output_item.added" in r.text
|
||||
assert "event: response.output_item.done" in r.text
|
||||
events = [line[len("event: ") :] for line in r.text.splitlines() if line.startswith("event: ")]
|
||||
assert "response.content_part.added" in events
|
||||
assert "response.output_text.done" in events
|
||||
assert "response.reasoning_text.delta" in events
|
||||
assert "response.reasoning_text.done" in events
|
||||
assert "response.function_call_arguments.delta" in events
|
||||
assert "response.function_call_arguments.done" in events
|
||||
content_part_added = _sse_payload(r.text, "response.content_part.added")
|
||||
assert content_part_added["part"] == {"annotations": [], "text": "", "type": "output_text"}
|
||||
added_items = [
|
||||
json.loads(line[len("data: ") :])["item"]
|
||||
for line in r.text.splitlines()
|
||||
if line.startswith("data: ") and '"type":"response.output_item.added"' in line
|
||||
]
|
||||
assert [item["type"] for item in added_items] == [
|
||||
"message",
|
||||
"reasoning",
|
||||
"function_call",
|
||||
"function_call_output",
|
||||
]
|
||||
assert added_items[0]["content"] == []
|
||||
assert added_items[1]["content"] == []
|
||||
assert added_items[2]["name"] == "lookup"
|
||||
assert added_items[2]["arguments"] == ""
|
||||
assert added_items[3]["output"] == [
|
||||
{"detail": "auto", "type": "input_image", "image_url": "https://example.com/cat.png"}
|
||||
]
|
||||
completed = _sse_payload(r.text, "response.completed")
|
||||
assert completed["response"]["output"][0]["content"][0]["text"] == "caption"
|
||||
assert completed["response"]["output"][1]["content"][0]["text"] == "thinking"
|
||||
assert completed["response"]["output"][2]["name"] == "lookup"
|
||||
assert completed["response"]["output"][3]["output"] == [
|
||||
{"detail": "auto", "type": "input_image", "image_url": "https://example.com/cat.png"}
|
||||
]
|
||||
|
||||
def test_sse_emits_failed_when_stream_raises(self) -> None:
|
||||
# Regression: ResponseOutputMessage.status only accepts in_progress/
|
||||
# completed/incomplete, so building an OpenAIResponse with status="failed"
|
||||
# used to crash with a pydantic ValidationError. The channel must map the
|
||||
# nested message status to "incomplete" while keeping the top-level
|
||||
# Response.status="failed".
|
||||
class _BoomStream:
|
||||
def __aiter__(self) -> AsyncIterator[AgentResponseUpdate]:
|
||||
async def _gen() -> AsyncIterator[AgentResponseUpdate]:
|
||||
yield AgentResponseUpdate(contents=[Content.from_text("partial")], role="assistant")
|
||||
raise RuntimeError("upstream blew up")
|
||||
|
||||
return _gen()
|
||||
|
||||
async def get_final_response(self) -> _FakeAgentResponse: # pragma: no cover
|
||||
return _FakeAgentResponse(text="")
|
||||
|
||||
class _BoomAgent(_FakeAgent):
|
||||
def run(self, messages: Any = None, *, stream: bool = False, **kwargs: Any) -> Any:
|
||||
self.calls.append({"messages": messages, "stream": stream, "kwargs": kwargs})
|
||||
if stream:
|
||||
return _BoomStream()
|
||||
raise AssertionError("non-streaming path not exercised here")
|
||||
|
||||
host = AgentFrameworkHost(target=_BoomAgent(), channels=[ResponsesChannel()])
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/responses", json={"input": "hi", "stream": True})
|
||||
assert r.status_code == 200
|
||||
body = r.text
|
||||
|
||||
events = [line[len("event: ") :] for line in body.splitlines() if line.startswith("event: ")]
|
||||
assert events[0] == "response.created"
|
||||
assert events[-1] == "response.failed"
|
||||
# The failed envelope must serialize cleanly — i.e. no ValidationError raised.
|
||||
assert "upstream blew up" in body
|
||||
@@ -1,169 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Tests for the OpenAI Responses request-body parser."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from agent_framework_hosting_responses import (
|
||||
messages_from_responses_input,
|
||||
parse_responses_identity,
|
||||
parse_responses_request,
|
||||
)
|
||||
|
||||
|
||||
class TestMessagesFromResponsesInput:
|
||||
def test_string_input_becomes_single_user_message(self) -> None:
|
||||
msgs = messages_from_responses_input("hello")
|
||||
assert len(msgs) == 1
|
||||
assert msgs[0].role == "user"
|
||||
assert msgs[0].text == "hello"
|
||||
|
||||
def test_input_text_items_collapse_into_one_user_message(self) -> None:
|
||||
msgs = messages_from_responses_input([{"type": "input_text", "text": "a"}, {"type": "input_text", "text": "b"}])
|
||||
assert len(msgs) == 1
|
||||
assert msgs[0].role == "user"
|
||||
assert msgs[0].text == "a b"
|
||||
|
||||
def test_message_envelope_with_string_content(self) -> None:
|
||||
msgs = messages_from_responses_input([
|
||||
{"type": "message", "role": "system", "content": "be brief"},
|
||||
{"type": "message", "role": "user", "content": "hi"},
|
||||
])
|
||||
assert [m.role for m in msgs] == ["system", "user"]
|
||||
assert msgs[0].text == "be brief"
|
||||
|
||||
def test_message_envelope_with_content_parts(self) -> None:
|
||||
msgs = messages_from_responses_input([
|
||||
{
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "describe this"}],
|
||||
}
|
||||
])
|
||||
assert msgs[0].text == "describe this"
|
||||
|
||||
def test_message_envelope_rejects_non_object_content_item(self) -> None:
|
||||
with pytest.raises(ValueError, match="content.*object"):
|
||||
messages_from_responses_input([{"type": "message", "role": "user", "content": ["bad"]}])
|
||||
|
||||
def test_message_envelope_rejects_invalid_content_shape(self) -> None:
|
||||
with pytest.raises(ValueError, match="content.*string or list"):
|
||||
messages_from_responses_input([{"type": "message", "role": "user", "content": 42}])
|
||||
|
||||
def test_input_file_via_url(self) -> None:
|
||||
msgs = messages_from_responses_input([
|
||||
{"type": "input_file", "file_url": "https://example.com/report.pdf", "mime_type": "application/pdf"}
|
||||
])
|
||||
assert msgs[0].contents[0].uri == "https://example.com/report.pdf"
|
||||
|
||||
def test_input_file_via_file_id(self) -> None:
|
||||
msgs = messages_from_responses_input([{"type": "input_file", "file_id": "file_123"}])
|
||||
assert msgs[0].contents[0].file_id == "file_123"
|
||||
|
||||
def test_input_file_missing_anchor_raises(self) -> None:
|
||||
with pytest.raises(ValueError, match="input_file"):
|
||||
messages_from_responses_input([{"type": "input_file"}])
|
||||
|
||||
def test_pending_text_flushes_before_message_envelope(self) -> None:
|
||||
msgs = messages_from_responses_input([
|
||||
{"type": "input_text", "text": "first"},
|
||||
{"type": "message", "role": "user", "content": "second"},
|
||||
])
|
||||
assert len(msgs) == 2
|
||||
assert msgs[0].text == "first"
|
||||
assert msgs[1].text == "second"
|
||||
|
||||
def test_image_url_via_string(self) -> None:
|
||||
msgs = messages_from_responses_input([{"type": "input_image", "image_url": "https://example.com/cat.png"}])
|
||||
assert len(msgs) == 1
|
||||
# Image content present.
|
||||
assert any(getattr(c, "uri", None) == "https://example.com/cat.png" for c in msgs[0].contents)
|
||||
|
||||
def test_image_url_via_object(self) -> None:
|
||||
msgs = messages_from_responses_input([
|
||||
{"type": "input_image", "image_url": {"url": "https://example.com/cat.png"}}
|
||||
])
|
||||
assert any(getattr(c, "uri", None) == "https://example.com/cat.png" for c in msgs[0].contents)
|
||||
|
||||
def test_unknown_input_type_raises(self) -> None:
|
||||
with pytest.raises(ValueError, match="Unsupported"):
|
||||
messages_from_responses_input([{"type": "weird"}])
|
||||
|
||||
def test_empty_list_raises(self) -> None:
|
||||
with pytest.raises(ValueError, match="non-empty"):
|
||||
messages_from_responses_input([])
|
||||
|
||||
def test_non_string_non_list_raises(self) -> None:
|
||||
with pytest.raises(ValueError):
|
||||
messages_from_responses_input(42) # type: ignore[arg-type]
|
||||
|
||||
def test_image_url_missing_raises(self) -> None:
|
||||
with pytest.raises(ValueError, match="image_url"):
|
||||
messages_from_responses_input([{"type": "input_image"}])
|
||||
|
||||
|
||||
class TestParseResponsesRequest:
|
||||
def test_known_fields_remapped_and_unknown_forwarded(self) -> None:
|
||||
_, opts, _ = parse_responses_request({
|
||||
"input": "hi",
|
||||
"instructions": "be brief",
|
||||
"temperature": 0.4,
|
||||
"top_p": 0.9,
|
||||
"tool_choice": "auto",
|
||||
"max_output_tokens": 256,
|
||||
"parallel_tool_calls": False,
|
||||
"truncation": "auto",
|
||||
"reasoning": {"effort": "low"},
|
||||
})
|
||||
# Known remaps applied.
|
||||
assert opts["max_tokens"] == 256
|
||||
assert opts["allow_multiple_tool_calls"] is False
|
||||
# Straight-through fields present.
|
||||
assert opts["temperature"] == 0.4
|
||||
assert opts["instructions"] == "be brief"
|
||||
assert opts["truncation"] == "auto"
|
||||
# Transport/session keys excluded.
|
||||
for key in ("input", "stream", "previous_response_id"):
|
||||
assert key not in opts
|
||||
|
||||
def test_model_passes_through_transport_keys_excluded(self) -> None:
|
||||
_, opts, _ = parse_responses_request({
|
||||
"input": "x",
|
||||
"model": "gpt-x",
|
||||
"stream": True,
|
||||
"previous_response_id": "r",
|
||||
})
|
||||
for key in ("input", "stream", "previous_response_id"):
|
||||
assert key not in opts
|
||||
# model passes through — not a transport key; run_hook decides what to do with it.
|
||||
assert opts["model"] == "gpt-x"
|
||||
|
||||
def test_none_values_dropped(self) -> None:
|
||||
_, opts, _ = parse_responses_request({"input": "x", "temperature": None})
|
||||
assert "temperature" not in opts
|
||||
|
||||
def test_previous_response_id_becomes_session(self) -> None:
|
||||
_, _, sess = parse_responses_request({"input": "x", "previous_response_id": "resp_42"})
|
||||
assert sess is not None
|
||||
assert sess.isolation_key == "resp_42"
|
||||
|
||||
|
||||
class TestParseResponsesIdentity:
|
||||
def test_safety_identifier_preferred(self) -> None:
|
||||
ident = parse_responses_identity({"safety_identifier": "abc", "user": "legacy"}, "responses")
|
||||
assert ident is not None
|
||||
assert ident.native_id == "abc"
|
||||
assert ident.channel == "responses"
|
||||
|
||||
def test_fallback_to_user(self) -> None:
|
||||
ident = parse_responses_identity({"user": "legacy"}, "responses")
|
||||
assert ident is not None
|
||||
assert ident.native_id == "legacy"
|
||||
|
||||
def test_returns_none_when_absent(self) -> None:
|
||||
assert parse_responses_identity({}, "responses") is None
|
||||
|
||||
def test_returns_none_for_non_string(self) -> None:
|
||||
assert parse_responses_identity({"safety_identifier": 42}, "responses") is None
|
||||
@@ -1,21 +0,0 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) Microsoft Corporation.
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE
|
||||
@@ -1,122 +0,0 @@
|
||||
# agent-framework-hosting
|
||||
|
||||
Multi-channel hosting for Microsoft Agent Framework agents.
|
||||
|
||||
`agent-framework-hosting` lets you serve a single agent or workflow target
|
||||
through one or more **channels**. The host owns one Starlette ASGI app,
|
||||
route/lifecycle composition, and per-`isolation_key` session resolution.
|
||||
Each channel owns its protocol parsing and response rendering.
|
||||
|
||||
The base package contains only channel-neutral plumbing:
|
||||
|
||||
- `AgentFrameworkHost` — the Starlette host.
|
||||
- `Channel` — the channel protocol.
|
||||
- `ChannelRequest` / `ChannelSession` / `ChannelIdentity` — the request
|
||||
envelope and optional channel metadata.
|
||||
- `ChannelContext` / `ChannelContribution` / `ChannelCommand` — channel-side
|
||||
hooks for invoking the target and contributing routes, commands, and
|
||||
lifecycle callbacks.
|
||||
- `ChannelRunHook` / `ChannelResponseHook` / `ChannelStreamUpdateHook` —
|
||||
host-invoked customization seams.
|
||||
|
||||
`ChannelStreamUpdateHook` applies to streamed updates only. It is not a
|
||||
substitute for final-response redaction.
|
||||
|
||||
Concrete channels live in their own packages so you only install what you use:
|
||||
|
||||
| Package | Transport |
|
||||
|---|---|
|
||||
| `agent-framework-hosting-responses` | OpenAI Responses API |
|
||||
|
||||
Additional channel packages can build on the same host contract without adding
|
||||
their protocol dependencies to the base package.
|
||||
|
||||
## Install
|
||||
|
||||
```bash
|
||||
pip install agent-framework-hosting agent-framework-hosting-responses
|
||||
# or with Hypercorn pre-installed for the demo `host.serve(...)` helper
|
||||
pip install "agent-framework-hosting[serve]" agent-framework-hosting-responses
|
||||
# add the [disk] extra to persist reset-session aliases
|
||||
pip install "agent-framework-hosting[disk]"
|
||||
```
|
||||
|
||||
## Quickstart
|
||||
|
||||
```python
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
from agent_framework_hosting import AgentFrameworkHost, Channel
|
||||
|
||||
agent = OpenAIChatClient().as_agent(name="Assistant")
|
||||
|
||||
# Add channels from sibling packages, e.g. `agent-framework-hosting-responses`
|
||||
# exposes a `ResponsesChannel` that serves the OpenAI Responses API.
|
||||
channels: list[Channel] = []
|
||||
|
||||
host = AgentFrameworkHost(target=agent, channels=channels)
|
||||
host.serve(port=8000)
|
||||
```
|
||||
|
||||
## Session state and workflow checkpoints
|
||||
|
||||
By default the host keeps live `AgentSession` objects and reset-session aliases
|
||||
in memory. Channels opt into continuity by setting
|
||||
`ChannelRequest.session = ChannelSession(isolation_key=...)`; requests with the
|
||||
same isolation key reuse the same host-created session.
|
||||
|
||||
The host treats `isolation_key` as an opaque partition key. Each channel or
|
||||
hosting environment decides where that key comes from:
|
||||
|
||||
- protocol headers supplied by a trusted platform,
|
||||
- request body fields such as a previous response or conversation ID,
|
||||
- route/path parameters,
|
||||
- channel-native metadata such as chat/user IDs, or
|
||||
- environment-provided context in an ephemeral host.
|
||||
|
||||
The host should be able to carry any of those sources as long as the channel or
|
||||
platform has already authenticated and authorized the caller before passing the
|
||||
key to `ChannelSession`.
|
||||
|
||||
The built-in request-context helper recognizes the `x-agent-user-isolation-key`
|
||||
and `x-agent-chat-isolation-key` header names because some hosting
|
||||
environments, including Foundry Hosted Agents, already use them. Reusing those
|
||||
header names does **not** mean `agent-framework-hosting` is the supported way to
|
||||
run on Foundry Hosted Agents; use `agent-framework-foundry-hosting` for that
|
||||
hosting surface.
|
||||
|
||||
For long-running deployments that need `reset_session(...)` aliases to survive
|
||||
restart, pass `state_dir`:
|
||||
|
||||
```python
|
||||
host = AgentFrameworkHost(
|
||||
target=agent,
|
||||
channels=channels,
|
||||
state_dir="./.host-state",
|
||||
)
|
||||
```
|
||||
|
||||
This creates `./.host-state/sessions/` and stores only lightweight alias
|
||||
bookkeeping. Live `AgentSession` objects are still rehydrated lazily by the
|
||||
configured history provider on the next turn.
|
||||
|
||||
For workflow targets, `checkpoint_location=...` is the clearest way to enable
|
||||
checkpoint persistence. As a convenience, `state_dir="./.host-state"` also
|
||||
derives `./.host-state/checkpoints/` for workflow targets. Use the mapping form
|
||||
when you want only one component:
|
||||
|
||||
```python
|
||||
from agent_framework_hosting import HostStatePaths
|
||||
|
||||
host = AgentFrameworkHost(
|
||||
target=workflow,
|
||||
channels=channels,
|
||||
state_dir=HostStatePaths(
|
||||
sessions="/var/lib/myapp/sessions",
|
||||
checkpoints="/var/lib/myapp/checkpoints",
|
||||
),
|
||||
)
|
||||
```
|
||||
|
||||
Cross-channel identity linking, multicast delivery, background runs,
|
||||
continuation tokens, and durable delivery runners are follow-up enhancements,
|
||||
not part of this v1 host contract.
|
||||
@@ -1,66 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Multi-channel hosting for Microsoft Agent Framework agents.
|
||||
|
||||
Serve a single agent target through one or more **channels** — pluggable
|
||||
adapters that expose the target over different transports. The base
|
||||
package contains only the channel-neutral plumbing; concrete channels
|
||||
ship in their own packages, such as ``agent-framework-hosting-responses``,
|
||||
so users install only what they need.
|
||||
"""
|
||||
|
||||
import importlib.metadata
|
||||
|
||||
from ._host import AgentFrameworkHost, ChannelContext, logger
|
||||
from ._isolation import (
|
||||
ISOLATION_HEADER_CHAT,
|
||||
ISOLATION_HEADER_USER,
|
||||
IsolationKeys,
|
||||
get_current_isolation_keys,
|
||||
reset_current_isolation_keys,
|
||||
set_current_isolation_keys,
|
||||
)
|
||||
from ._types import (
|
||||
Channel,
|
||||
ChannelCommand,
|
||||
ChannelCommandContext,
|
||||
ChannelContribution,
|
||||
ChannelIdentity,
|
||||
ChannelRequest,
|
||||
ChannelResponseHook,
|
||||
ChannelRunHook,
|
||||
ChannelSession,
|
||||
ChannelStreamUpdateHook,
|
||||
HostedRunResult,
|
||||
HostStatePaths,
|
||||
)
|
||||
|
||||
try:
|
||||
__version__ = importlib.metadata.version(__name__)
|
||||
except importlib.metadata.PackageNotFoundError:
|
||||
__version__ = "0.0.0"
|
||||
|
||||
__all__ = [
|
||||
"ISOLATION_HEADER_CHAT",
|
||||
"ISOLATION_HEADER_USER",
|
||||
"AgentFrameworkHost",
|
||||
"Channel",
|
||||
"ChannelCommand",
|
||||
"ChannelCommandContext",
|
||||
"ChannelContext",
|
||||
"ChannelContribution",
|
||||
"ChannelIdentity",
|
||||
"ChannelRequest",
|
||||
"ChannelResponseHook",
|
||||
"ChannelRunHook",
|
||||
"ChannelSession",
|
||||
"ChannelStreamUpdateHook",
|
||||
"HostStatePaths",
|
||||
"HostedRunResult",
|
||||
"IsolationKeys",
|
||||
"__version__",
|
||||
"get_current_isolation_keys",
|
||||
"logger",
|
||||
"reset_current_isolation_keys",
|
||||
"set_current_isolation_keys",
|
||||
]
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,82 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Per-request isolation keys for host/platform-provided request context.
|
||||
|
||||
``ChannelSession.isolation_key`` is the host's generic session partition key,
|
||||
but different channels and platforms discover that key from different places:
|
||||
protocol headers, request bodies, URL/path segments, webhook metadata, or
|
||||
environment-provided context for ephemeral hosts.
|
||||
|
||||
This module covers the request-context case where a platform provides
|
||||
isolation outside the channel payload. The Foundry Hosted Agents runtime, for
|
||||
example, injects two well-known headers on requests it forwards to the user's
|
||||
container:
|
||||
|
||||
* ``x-agent-user-isolation-key`` — opaque per-user partition key
|
||||
* ``x-agent-chat-isolation-key`` — opaque per-conversation partition key
|
||||
|
||||
The generic host intentionally reuses those header names so the same isolation
|
||||
context can be consumed by supporting providers. Reusing the names does **not**
|
||||
mean ``agent-framework-hosting`` is a supported way to run on Foundry Hosted
|
||||
Agents; use ``agent-framework-foundry-hosting`` for that hosting surface.
|
||||
|
||||
When those headers are present the host-installed ASGI middleware pushes them
|
||||
into :data:`current_isolation_keys` for the duration of the request, then
|
||||
resets it. Channels may still choose a different session key source and pass it
|
||||
directly via ``ChannelSession(isolation_key=...)``.
|
||||
|
||||
The contextvar holds a plain :class:`IsolationKeys` mapping; conversion to
|
||||
provider-specific types happens at the consuming provider so this module has no
|
||||
provider dependencies.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from contextvars import ContextVar, Token
|
||||
|
||||
__all__ = [
|
||||
"ISOLATION_HEADER_CHAT",
|
||||
"ISOLATION_HEADER_USER",
|
||||
"IsolationKeys",
|
||||
"current_isolation_keys",
|
||||
"get_current_isolation_keys",
|
||||
"reset_current_isolation_keys",
|
||||
"set_current_isolation_keys",
|
||||
]
|
||||
|
||||
|
||||
ISOLATION_HEADER_USER = "x-agent-user-isolation-key"
|
||||
ISOLATION_HEADER_CHAT = "x-agent-chat-isolation-key"
|
||||
|
||||
|
||||
class IsolationKeys:
|
||||
"""Per-request isolation keys lifted from host/platform context."""
|
||||
|
||||
def __init__(self, user_key: str | None = None, chat_key: str | None = None) -> None:
|
||||
self.user_key = user_key
|
||||
self.chat_key = chat_key
|
||||
|
||||
@property
|
||||
def is_empty(self) -> bool:
|
||||
return self.user_key is None and self.chat_key is None
|
||||
|
||||
|
||||
current_isolation_keys: ContextVar[IsolationKeys | None] = ContextVar(
|
||||
"agent_framework_hosting_isolation_keys",
|
||||
default=None,
|
||||
)
|
||||
|
||||
|
||||
def get_current_isolation_keys() -> IsolationKeys | None:
|
||||
"""Return the isolation keys bound to the current request, if any."""
|
||||
return current_isolation_keys.get()
|
||||
|
||||
|
||||
def set_current_isolation_keys(keys: IsolationKeys | None) -> Token[IsolationKeys | None]:
|
||||
"""Bind ``keys`` to the current async context and return a reset token."""
|
||||
return current_isolation_keys.set(keys)
|
||||
|
||||
|
||||
def reset_current_isolation_keys(token: Token[IsolationKeys | None]) -> None:
|
||||
"""Restore the isolation contextvar to its prior value."""
|
||||
current_isolation_keys.reset(token)
|
||||
@@ -1,128 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Shared persistence primitives for the hosting package.
|
||||
|
||||
The simplified hosting core keeps disk persistence only for session aliases
|
||||
created by :meth:`AgentFrameworkHost.reset_session` and for workflow
|
||||
checkpoint path derivation. The on-disk session-alias store uses the optional
|
||||
``diskcache`` package installed via the ``[disk]`` extra.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import importlib
|
||||
import os
|
||||
import sys
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ._types import HostStatePaths
|
||||
|
||||
_KNOWN_COMPONENTS: tuple[str, ...] = ("sessions", "checkpoints")
|
||||
|
||||
|
||||
def load_diskcache() -> Any:
|
||||
"""Lazy-import :mod:`diskcache` with a helpful error when missing."""
|
||||
try:
|
||||
return importlib.import_module("diskcache")
|
||||
except ImportError as exc: # pragma: no cover - exercised via tests by monkeypatching
|
||||
raise ImportError(
|
||||
"agent-framework-hosting was asked to persist session aliases to disk "
|
||||
"(state_dir['sessions'] is set) but the optional `diskcache` dependency "
|
||||
"is not installed. Install the disk extra: "
|
||||
"`pip install 'agent-framework-hosting[disk]`."
|
||||
) from exc
|
||||
|
||||
|
||||
def acquire_state_dir_lock(component_dir: Path) -> Any:
|
||||
"""Acquire an exclusive single-owner lock on a component's state dir.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If another process already holds the lock.
|
||||
"""
|
||||
component_dir.mkdir(parents=True, exist_ok=True)
|
||||
lock_path = component_dir / ".lock"
|
||||
fh = open(lock_path, "a+", encoding="utf-8") # noqa: SIM115 - kept open for lifetime
|
||||
try:
|
||||
if sys.platform == "win32":
|
||||
import msvcrt
|
||||
|
||||
try:
|
||||
msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1)
|
||||
except OSError as exc:
|
||||
fh.close()
|
||||
raise RuntimeError(
|
||||
f"Another process already holds the hosting state lock at {lock_path}. "
|
||||
"Point each host at its own state_dir."
|
||||
) from exc
|
||||
else:
|
||||
import fcntl
|
||||
|
||||
try:
|
||||
fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except OSError as exc:
|
||||
fh.close()
|
||||
raise RuntimeError(
|
||||
f"Another process already holds the hosting state lock at {lock_path}. "
|
||||
"Point each host at its own state_dir."
|
||||
) from exc
|
||||
except RuntimeError:
|
||||
raise
|
||||
except Exception:
|
||||
fh.close()
|
||||
raise
|
||||
return fh
|
||||
|
||||
|
||||
def release_state_dir_lock(handle: Any) -> None:
|
||||
"""Release a lock previously acquired by :func:`acquire_state_dir_lock`."""
|
||||
if handle is None:
|
||||
return
|
||||
with contextlib.suppress(Exception):
|
||||
handle.close()
|
||||
|
||||
|
||||
def normalize_state_dir(
|
||||
state_dir: str | os.PathLike[str] | HostStatePaths | Mapping[str, str | os.PathLike[str]] | None,
|
||||
) -> dict[str, Path | None]:
|
||||
"""Resolve the host-level ``state_dir`` parameter into a per-component map.
|
||||
|
||||
Accepts ``None``, a single root path, or a mapping with ``sessions`` and
|
||||
``checkpoints`` keys. Unknown keys raise ``ValueError`` so obsolete
|
||||
``runner`` / ``links`` configuration is rejected instead of silently
|
||||
doing nothing.
|
||||
"""
|
||||
result: dict[str, Path | None] = {name: None for name in _KNOWN_COMPONENTS}
|
||||
if state_dir is None:
|
||||
return result
|
||||
|
||||
if isinstance(state_dir, (str, os.PathLike)):
|
||||
root = Path(os.fspath(state_dir))
|
||||
for name in _KNOWN_COMPONENTS:
|
||||
result[name] = root / name
|
||||
return result
|
||||
|
||||
if isinstance(state_dir, Mapping):
|
||||
unknown = [k for k in state_dir if k not in _KNOWN_COMPONENTS]
|
||||
if unknown:
|
||||
raise ValueError(
|
||||
f"state_dir mapping contains unknown component key(s): {unknown!r}. "
|
||||
f"Known components are: {list(_KNOWN_COMPONENTS)!r}."
|
||||
)
|
||||
for name in _KNOWN_COMPONENTS:
|
||||
raw_value: Any = state_dir.get(name)
|
||||
if raw_value is None:
|
||||
result[name] = None
|
||||
continue
|
||||
if isinstance(raw_value, (str, os.PathLike)):
|
||||
result[name] = Path(os.fspath(raw_value))
|
||||
else:
|
||||
raise TypeError(f"state_dir[{name!r}] must be a str or PathLike — got {type(raw_value).__name__}")
|
||||
return result
|
||||
|
||||
raise TypeError(
|
||||
f"state_dir must be a str, PathLike, HostStatePaths mapping, or None — got {type(state_dir).__name__}"
|
||||
)
|
||||
@@ -1,146 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Disk-backed wrapper for the host's session-alias map.
|
||||
|
||||
``AgentFrameworkHost.reset_session(isolation_key)`` rotates future requests for
|
||||
that isolation key onto a new session id. Persisting the alias map lets that
|
||||
rotation survive a host restart without introducing cross-channel identity or
|
||||
delivery state into the core host.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from typing import Any, TypeVar
|
||||
|
||||
from ._persistence import (
|
||||
acquire_state_dir_lock,
|
||||
load_diskcache,
|
||||
release_state_dir_lock,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_V = TypeVar("_V")
|
||||
_ALIASES_PREFIX = "aliases:"
|
||||
|
||||
|
||||
class SessionsStateStore:
|
||||
"""One disk cache + lock for host-side session aliases."""
|
||||
|
||||
def __init__(self, sessions_dir: str | os.PathLike[str]) -> None:
|
||||
self._sessions_dir: Path = Path(os.fspath(sessions_dir))
|
||||
diskcache = load_diskcache()
|
||||
self._lock_handle: Any = acquire_state_dir_lock(self._sessions_dir)
|
||||
try:
|
||||
self._cache: Any = diskcache.Cache(str(self._sessions_dir))
|
||||
except Exception:
|
||||
release_state_dir_lock(self._lock_handle)
|
||||
self._lock_handle = None
|
||||
raise
|
||||
|
||||
@property
|
||||
def cache(self) -> Any:
|
||||
"""Return the underlying :mod:`diskcache` Cache."""
|
||||
return self._cache
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close the cache and release the directory lock."""
|
||||
if self._cache is not None:
|
||||
try:
|
||||
self._cache.close()
|
||||
except Exception: # pragma: no cover - close errors aren't actionable
|
||||
logger.exception("SessionsStateStore: failed to close cache cleanly")
|
||||
self._cache = None
|
||||
if self._lock_handle is not None:
|
||||
release_state_dir_lock(self._lock_handle)
|
||||
self._lock_handle = None
|
||||
|
||||
|
||||
class _PersistedDict(dict[str, _V]):
|
||||
"""Drop-in :class:`dict` whose mutations mirror to a diskcache prefix."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
store: SessionsStateStore,
|
||||
key_prefix: str,
|
||||
initial: Mapping[str, _V] | None = None,
|
||||
) -> None:
|
||||
super().__init__()
|
||||
self._store = store
|
||||
self._prefix = key_prefix
|
||||
cache: Any = store.cache
|
||||
for raw_key in cache.iterkeys():
|
||||
if not isinstance(raw_key, str) or not raw_key.startswith(key_prefix):
|
||||
continue
|
||||
try:
|
||||
value: Any = cache.get(raw_key)
|
||||
except Exception:
|
||||
logger.exception("SessionsStateStore: failed to rehydrate %s; skipping", raw_key)
|
||||
continue
|
||||
logical_key = raw_key[len(key_prefix) :]
|
||||
super().__setitem__(logical_key, value)
|
||||
if initial:
|
||||
for key, value in initial.items():
|
||||
self[key] = value
|
||||
|
||||
def __setitem__(self, key: str, value: _V) -> None:
|
||||
super().__setitem__(key, value)
|
||||
try:
|
||||
self._store.cache.set(self._prefix + key, value)
|
||||
except Exception: # pragma: no cover - cache write failures aren't actionable
|
||||
logger.exception("SessionsStateStore: failed to persist %s%s", self._prefix, key)
|
||||
|
||||
def __delitem__(self, key: str) -> None:
|
||||
super().__delitem__(key)
|
||||
try:
|
||||
del self._store.cache[self._prefix + key]
|
||||
except KeyError:
|
||||
pass
|
||||
except Exception: # pragma: no cover - cache write failures aren't actionable
|
||||
logger.exception("SessionsStateStore: failed to evict %s%s", self._prefix, key)
|
||||
|
||||
def pop(self, key: str, *args: Any) -> _V:
|
||||
"""Mirror ``dict.pop`` to disk."""
|
||||
value: _V = super().pop(key, *args)
|
||||
try:
|
||||
del self._store.cache[self._prefix + key]
|
||||
except KeyError:
|
||||
pass
|
||||
except Exception: # pragma: no cover
|
||||
logger.exception("SessionsStateStore: failed to evict %s%s", self._prefix, key)
|
||||
return value
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Mirror ``dict.clear`` to disk."""
|
||||
keys = list(self.keys())
|
||||
super().clear()
|
||||
cache = self._store.cache
|
||||
for key in keys:
|
||||
try:
|
||||
del cache[self._prefix + key]
|
||||
except KeyError:
|
||||
pass
|
||||
except Exception: # pragma: no cover
|
||||
logger.exception("SessionsStateStore: failed to evict %s%s during clear", self._prefix, key)
|
||||
|
||||
def update( # type: ignore[override]
|
||||
self,
|
||||
other: Mapping[str, _V] | None = None,
|
||||
/,
|
||||
**kwargs: _V,
|
||||
) -> None:
|
||||
"""Mirror ``dict.update`` to disk one item at a time."""
|
||||
if other is not None:
|
||||
for key in other:
|
||||
self[key] = other[key]
|
||||
for key, value in kwargs.items():
|
||||
self[key] = value
|
||||
|
||||
|
||||
def build_session_aliases(store: SessionsStateStore) -> dict[str, str]:
|
||||
"""Return the disk-backed session-alias map for ``store``."""
|
||||
return _PersistedDict[str](store, _ALIASES_PREFIX)
|
||||
@@ -1,212 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
# ``ChannelRequest`` is the only intentional dataclass here (callers use
|
||||
# ``dataclasses.replace`` on it in run hooks). The other types are plain
|
||||
# Python classes by preference, so the "could be a dataclass" lint is muted
|
||||
# at the file level.
|
||||
# ruff: noqa: B903
|
||||
|
||||
"""Channel-neutral request envelope and channel protocol types.
|
||||
|
||||
These types form the boundary between the host and individual channels.
|
||||
A channel parses its native payload, builds a :class:`ChannelRequest`, and
|
||||
hands it to :class:`ChannelContext.run` (or ``run_stream``) on the host.
|
||||
The channel owns rendering the result back onto its originating protocol.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from collections.abc import Awaitable, Callable, Mapping, Sequence
|
||||
from dataclasses import dataclass, field
|
||||
from typing import TYPE_CHECKING, Any, Generic, Protocol, TypedDict, TypeVar, runtime_checkable
|
||||
|
||||
from agent_framework import (
|
||||
AgentResponseUpdate,
|
||||
AgentRunInputs,
|
||||
)
|
||||
from starlette.routing import BaseRoute
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ._host import ChannelContext
|
||||
|
||||
|
||||
class ChannelSession:
|
||||
"""Channel-supplied session hint.
|
||||
|
||||
The host turns this into an ``AgentSession`` keyed by ``isolation_key`` so
|
||||
every distinct end user gets their own context-provider state (e.g. one
|
||||
``FileHistoryProvider`` JSONL file per user).
|
||||
"""
|
||||
|
||||
def __init__(self, isolation_key: str | None = None) -> None:
|
||||
self.isolation_key = isolation_key
|
||||
|
||||
|
||||
class ChannelIdentity:
|
||||
"""Channel-native identity metadata observed on a request.
|
||||
|
||||
The simplified hosting core records this only on the persisted input
|
||||
message's ``additional_properties["hosting"]`` block and forwards it
|
||||
through run/response hooks. Cross-channel linking and recipient lookup are
|
||||
follow-up concerns, not part of the v1 host contract.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
channel: str,
|
||||
native_id: str,
|
||||
attributes: Mapping[str, Any] | None = None,
|
||||
) -> None:
|
||||
self.channel = channel
|
||||
self.native_id = native_id
|
||||
self.attributes: Mapping[str, Any] = attributes if attributes is not None else dict()
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelRequest:
|
||||
"""Uniform invocation envelope every channel produces from its native payload.
|
||||
|
||||
Kept as a dataclass so app authors can use ``dataclasses.replace(...)`` in
|
||||
run hooks to produce a modified envelope without re-listing every field.
|
||||
"""
|
||||
|
||||
channel: str
|
||||
operation: str
|
||||
input: AgentRunInputs
|
||||
session: ChannelSession | None = None
|
||||
options: Mapping[str, Any] | None = None
|
||||
session_mode: str = "auto"
|
||||
metadata: Mapping[str, Any] = field(default_factory=lambda: {})
|
||||
attributes: Mapping[str, Any] = field(default_factory=lambda: {})
|
||||
stream: bool = False
|
||||
identity: ChannelIdentity | None = None
|
||||
|
||||
|
||||
class ChannelCommand:
|
||||
"""A discoverable command a channel exposes to its users (e.g. ``/reset``)."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
description: str,
|
||||
handle: Callable[[ChannelCommandContext], Awaitable[None]],
|
||||
) -> None:
|
||||
self.name = name
|
||||
self.description = description
|
||||
self.handle = handle
|
||||
|
||||
|
||||
class ChannelCommandContext:
|
||||
"""Context passed to a :class:`ChannelCommand` handler."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
request: ChannelRequest,
|
||||
reply: Callable[[str], Awaitable[None]],
|
||||
) -> None:
|
||||
self.request = request
|
||||
self.reply = reply
|
||||
|
||||
|
||||
_EMPTY_ROUTES: tuple[BaseRoute, ...] = ()
|
||||
_EMPTY_COMMANDS: tuple[ChannelCommand, ...] = ()
|
||||
_EMPTY_LIFECYCLE: tuple[Callable[[], Awaitable[None]], ...] = ()
|
||||
|
||||
|
||||
class ChannelContribution:
|
||||
"""Routes, commands, and lifecycle hooks a channel contributes to the host."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
routes: Sequence[BaseRoute] = _EMPTY_ROUTES,
|
||||
commands: Sequence[ChannelCommand] = _EMPTY_COMMANDS,
|
||||
on_startup: Sequence[Callable[[], Awaitable[None]]] = _EMPTY_LIFECYCLE,
|
||||
on_shutdown: Sequence[Callable[[], Awaitable[None]]] = _EMPTY_LIFECYCLE,
|
||||
) -> None:
|
||||
self.routes = routes
|
||||
self.commands = commands
|
||||
self.on_startup = on_startup
|
||||
self.on_shutdown = on_shutdown
|
||||
|
||||
|
||||
class _Unset:
|
||||
"""Sentinel for ``HostedRunResult.replace`` overrides.
|
||||
|
||||
Distinguishes "caller did not pass this kwarg" from "caller passed
|
||||
``None`` explicitly" — needed because ``session`` is ``None`` in
|
||||
many envelopes and we want the no-arg call to preserve it.
|
||||
"""
|
||||
|
||||
|
||||
_UNSET = _Unset()
|
||||
|
||||
|
||||
TResult = TypeVar("TResult")
|
||||
|
||||
|
||||
class HostedRunResult(Generic[TResult]):
|
||||
"""Channel-neutral envelope around the target's full-fidelity result.
|
||||
|
||||
The host does not flatten or pre-shape the target output. Channels and
|
||||
response hooks read the underlying result type directly and serialize the
|
||||
subset their wire format can carry.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
result: TResult,
|
||||
*,
|
||||
session: Any | None = None,
|
||||
) -> None:
|
||||
self.result = result
|
||||
self.session = session
|
||||
|
||||
def replace(
|
||||
self,
|
||||
*,
|
||||
result: TResult | _Unset = _UNSET,
|
||||
session: Any | _Unset | None = _UNSET,
|
||||
) -> HostedRunResult[TResult]:
|
||||
"""Return a shallow copy with the supplied fields overridden."""
|
||||
new: HostedRunResult[TResult] = HostedRunResult.__new__(HostedRunResult) # pyright: ignore[reportUnknownVariableType]
|
||||
new.result = self.result if isinstance(result, _Unset) else result
|
||||
new.session = self.session if isinstance(session, _Unset) else session
|
||||
return new
|
||||
|
||||
|
||||
class HostStatePaths(TypedDict, total=False):
|
||||
"""Per-component disk paths for host-managed state.
|
||||
|
||||
Only session aliases and workflow checkpoints remain in the simplified
|
||||
host. Linking stores, active-channel maps, identity registries, and runner
|
||||
queues are follow-up concerns.
|
||||
"""
|
||||
|
||||
sessions: str | os.PathLike[str]
|
||||
"""Where the host persists session aliases created by ``reset_session``."""
|
||||
|
||||
checkpoints: str | os.PathLike[str]
|
||||
"""Where the host persists workflow checkpoints for ``Workflow`` targets."""
|
||||
|
||||
|
||||
ChannelStreamUpdateHook = Callable[
|
||||
[AgentResponseUpdate],
|
||||
"AgentResponseUpdate | Awaitable[AgentResponseUpdate | None] | None",
|
||||
]
|
||||
|
||||
|
||||
ChannelRunHook = Callable[..., "Awaitable[ChannelRequest] | ChannelRequest"]
|
||||
|
||||
|
||||
ChannelResponseHook = Callable[..., "Awaitable[HostedRunResult[Any]] | HostedRunResult[Any]"]
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class Channel(Protocol):
|
||||
"""A pluggable adapter that exposes one transport on the host."""
|
||||
|
||||
name: str
|
||||
path: str
|
||||
|
||||
def contribute(self, context: ChannelContext) -> ChannelContribution: ...
|
||||
@@ -1,92 +0,0 @@
|
||||
[project]
|
||||
name = "agent-framework-hosting"
|
||||
description = "Multi-channel hosting for Microsoft Agent Framework agents."
|
||||
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.10"
|
||||
version = "1.0.0a260424"
|
||||
license-files = ["LICENSE"]
|
||||
urls.homepage = "https://aka.ms/agent-framework"
|
||||
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
|
||||
urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true"
|
||||
urls.issues = "https://github.com/microsoft/agent-framework/issues"
|
||||
classifiers = [
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Development Status :: 3 - Alpha",
|
||||
"Intended Audience :: Developers",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Programming Language :: Python :: 3.13",
|
||||
"Programming Language :: Python :: 3.14",
|
||||
"Typing :: Typed",
|
||||
]
|
||||
dependencies = [
|
||||
"agent-framework-core>=1.2.0,<2",
|
||||
"starlette>=0.37",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
serve = [
|
||||
"hypercorn>=0.17",
|
||||
]
|
||||
disk = [
|
||||
"diskcache>=5.6",
|
||||
]
|
||||
|
||||
[tool.uv]
|
||||
prerelease = "if-necessary-or-explicit"
|
||||
environments = [
|
||||
"sys_platform == 'darwin'",
|
||||
"sys_platform == 'linux'",
|
||||
"sys_platform == 'win32'"
|
||||
]
|
||||
|
||||
[tool.uv-dynamic-versioning]
|
||||
fallback-version = "0.0.0"
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = 'tests'
|
||||
addopts = "-ra -q -r fEX"
|
||||
asyncio_mode = "auto"
|
||||
asyncio_default_fixture_loop_scope = "function"
|
||||
filterwarnings = []
|
||||
timeout = 120
|
||||
markers = [
|
||||
"integration: marks tests as integration tests that require external services",
|
||||
]
|
||||
|
||||
[tool.ruff]
|
||||
extend = "../../pyproject.toml"
|
||||
|
||||
[tool.coverage.run]
|
||||
omit = [
|
||||
"**/__init__.py"
|
||||
]
|
||||
|
||||
[tool.pyright]
|
||||
extends = "../../pyproject.toml"
|
||||
include = ["agent_framework_hosting"]
|
||||
exclude = ['tests']
|
||||
|
||||
[tool.bandit]
|
||||
targets = ["agent_framework_hosting"]
|
||||
exclude_dirs = ["tests"]
|
||||
|
||||
[tool.poe]
|
||||
executor.type = "uv"
|
||||
include = "../../shared_tasks.toml"
|
||||
|
||||
[tool.poe.tasks.test]
|
||||
help = "Run the default unit test suite for this package."
|
||||
cmd = 'pytest -m "not integration" --cov=agent_framework_hosting --cov-report=term-missing:skip-covered tests'
|
||||
|
||||
[build-system]
|
||||
requires = ["flit-core >= 3.11,<4.0"]
|
||||
build-backend = "flit_core.buildapi"
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"httpx>=0.28.1",
|
||||
]
|
||||
@@ -1,45 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Workflow fixtures for hosting tests.
|
||||
|
||||
Defined in a module that does not use ``from __future__ import annotations``
|
||||
because the workflow handler validation reflects on real annotation objects
|
||||
rather than stringified forms.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from agent_framework import Executor, Workflow, WorkflowBuilder, WorkflowContext, handler
|
||||
|
||||
|
||||
class _UpperExecutor(Executor):
|
||||
@handler
|
||||
async def handle(self, text: str, ctx: WorkflowContext[Any, str]) -> None:
|
||||
await ctx.yield_output(text.upper())
|
||||
|
||||
|
||||
class _EchoExecutor(Executor):
|
||||
@handler
|
||||
async def handle(self, text: str, ctx: WorkflowContext[Any, str]) -> None:
|
||||
await ctx.yield_output(text)
|
||||
|
||||
|
||||
def build_upper_workflow() -> Workflow:
|
||||
return WorkflowBuilder(start_executor=_UpperExecutor(id="upper")).build()
|
||||
|
||||
|
||||
def build_echo_workflow() -> Workflow:
|
||||
return WorkflowBuilder(start_executor=_EchoExecutor(id="echo")).build()
|
||||
|
||||
|
||||
class _MultiChunkExecutor(Executor):
|
||||
"""Yields three separate ``output`` events so streaming has something to chew on."""
|
||||
|
||||
@handler
|
||||
async def handle(self, text: str, ctx: WorkflowContext[Any, str]) -> None:
|
||||
for chunk in (f"{text}-1", f"{text}-2", f"{text}-3"):
|
||||
await ctx.yield_output(chunk)
|
||||
|
||||
|
||||
def build_multi_chunk_workflow() -> Workflow:
|
||||
return WorkflowBuilder(start_executor=_MultiChunkExecutor(id="multi")).build()
|
||||
@@ -1,25 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Pytest configuration for hosting tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def pytest_configure() -> None:
|
||||
"""Make workflow fixtures importable in package-local and aggregate test modes."""
|
||||
module_name = "hosting_workflow_fixtures"
|
||||
if module_name in sys.modules:
|
||||
return
|
||||
|
||||
fixture_path = Path(__file__).with_name("_workflow_fixtures.py")
|
||||
spec = importlib.util.spec_from_file_location(module_name, fixture_path)
|
||||
if spec is None or spec.loader is None:
|
||||
raise ImportError(f"Unable to load workflow fixtures from {fixture_path}")
|
||||
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[module_name] = module
|
||||
spec.loader.exec_module(module)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,239 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Tests for narrowed ``state_dir`` support in :class:`AgentFrameworkHost`."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
from agent_framework import AgentSession
|
||||
|
||||
from agent_framework_hosting import AgentFrameworkHost, ChannelContext, ChannelContribution
|
||||
|
||||
pytest.importorskip("diskcache")
|
||||
|
||||
|
||||
class _AgentStub:
|
||||
"""Bare-minimum SupportsAgentRun stub for host construction."""
|
||||
|
||||
id = "agent-stub"
|
||||
name: str | None = "Agent Stub"
|
||||
description: str | None = "Test agent stub"
|
||||
|
||||
def create_session(self, *, session_id: str | None = None) -> AgentSession:
|
||||
return AgentSession(session_id=session_id)
|
||||
|
||||
def get_session(self, service_session_id: str, *, session_id: str | None = None) -> AgentSession:
|
||||
return AgentSession(service_session_id=service_session_id, session_id=session_id)
|
||||
|
||||
def run(self, *_args: Any, **_kwargs: Any) -> Any: # pragma: no cover - unused
|
||||
raise RuntimeError("not invoked")
|
||||
|
||||
|
||||
class _ChannelStub:
|
||||
name = "stub"
|
||||
path = "/stub"
|
||||
|
||||
def contribute(self, context: ChannelContext) -> ChannelContribution:
|
||||
del context
|
||||
return ChannelContribution()
|
||||
|
||||
|
||||
def _close_host_disk(host: AgentFrameworkHost) -> None:
|
||||
"""Release any session-alias store held by ``host``."""
|
||||
if host._sessions_store is not None:
|
||||
host._sessions_store.close()
|
||||
|
||||
|
||||
def test_state_dir_none_keeps_plain_alias_dict(tmp_path: Path) -> None:
|
||||
"""No store, no alias persistence, no files written."""
|
||||
host = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()])
|
||||
assert host._sessions_store is None
|
||||
assert isinstance(host._session_aliases, dict)
|
||||
assert list(tmp_path.iterdir()) == []
|
||||
|
||||
|
||||
def test_string_state_dir_creates_sessions_subfolder_only(tmp_path: Path) -> None:
|
||||
"""Passing a single path expands to ``sessions/`` plus lazy checkpoint path."""
|
||||
host = AgentFrameworkHost(
|
||||
target=_AgentStub(),
|
||||
channels=[_ChannelStub()],
|
||||
state_dir=tmp_path,
|
||||
)
|
||||
try:
|
||||
assert host._sessions_store is not None
|
||||
assert (tmp_path / "sessions").is_dir()
|
||||
assert not (tmp_path / "runner").exists()
|
||||
assert not (tmp_path / "links").exists()
|
||||
# Checkpoint path is derived but not created for agent targets.
|
||||
assert not (tmp_path / "checkpoints").exists()
|
||||
finally:
|
||||
_close_host_disk(host)
|
||||
|
||||
|
||||
def test_per_component_session_path(tmp_path: Path) -> None:
|
||||
"""Dict form lets callers route session aliases to a specific root."""
|
||||
sessions_dir = tmp_path / "state"
|
||||
host = AgentFrameworkHost(
|
||||
target=_AgentStub(),
|
||||
channels=[_ChannelStub()],
|
||||
state_dir={"sessions": sessions_dir},
|
||||
)
|
||||
try:
|
||||
assert sessions_dir.is_dir()
|
||||
assert host._sessions_store is not None
|
||||
assert host._checkpoint_location is None
|
||||
finally:
|
||||
_close_host_disk(host)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("key", ["runner", "links", "active", "identities"])
|
||||
def test_removed_state_dir_component_keys_raise(tmp_path: Path, key: str) -> None:
|
||||
"""Obsolete follow-up components should fail loudly instead of becoming no-ops."""
|
||||
with pytest.raises(ValueError, match="unknown"):
|
||||
AgentFrameworkHost(
|
||||
target=_AgentStub(),
|
||||
channels=[_ChannelStub()],
|
||||
state_dir=cast(Any, {key: tmp_path / key}),
|
||||
)
|
||||
|
||||
|
||||
def test_session_aliases_survive_restart(tmp_path: Path) -> None:
|
||||
"""Aliases written on host #1 must be visible to host #2."""
|
||||
state_dir = tmp_path / "state"
|
||||
|
||||
host1 = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()], state_dir=state_dir)
|
||||
host1._session_aliases["user-1"] = "sess-abc"
|
||||
host1._session_aliases["user-2"] = "sess-def"
|
||||
_close_host_disk(host1)
|
||||
|
||||
host2 = AgentFrameworkHost(target=_AgentStub(), channels=[_ChannelStub()], state_dir=state_dir)
|
||||
try:
|
||||
assert host2._session_aliases["user-1"] == "sess-abc"
|
||||
assert host2._session_aliases["user-2"] == "sess-def"
|
||||
finally:
|
||||
_close_host_disk(host2)
|
||||
|
||||
|
||||
def _build_simple_workflow() -> Any:
|
||||
"""Build a no-op workflow for checkpoint-wiring tests."""
|
||||
build_upper_workflow = importlib.import_module("hosting_workflow_fixtures").build_upper_workflow
|
||||
|
||||
return build_upper_workflow()
|
||||
|
||||
|
||||
def test_single_path_state_dir_wires_workflow_checkpoints(tmp_path: Path) -> None:
|
||||
"""``state_dir="/foo"`` + workflow target → ``/foo/checkpoints/`` is used."""
|
||||
workflow = _build_simple_workflow()
|
||||
host = AgentFrameworkHost(
|
||||
target=workflow,
|
||||
channels=[_ChannelStub()],
|
||||
state_dir=tmp_path,
|
||||
)
|
||||
try:
|
||||
assert host._checkpoint_location == tmp_path / "checkpoints"
|
||||
finally:
|
||||
_close_host_disk(host)
|
||||
|
||||
|
||||
def test_mapping_state_dir_checkpoints_key_wires_workflow_checkpoints(tmp_path: Path) -> None:
|
||||
"""``state_dir={"checkpoints": ...}`` + workflow target → that path is used."""
|
||||
workflow = _build_simple_workflow()
|
||||
ckpt_dir = tmp_path / "ck"
|
||||
host = AgentFrameworkHost(
|
||||
target=workflow,
|
||||
channels=[_ChannelStub()],
|
||||
state_dir={"checkpoints": ckpt_dir},
|
||||
)
|
||||
try:
|
||||
assert host._checkpoint_location == ckpt_dir
|
||||
assert host._sessions_store is None
|
||||
finally:
|
||||
_close_host_disk(host)
|
||||
|
||||
|
||||
def test_mapping_state_dir_omits_checkpoints_for_workflow(tmp_path: Path) -> None:
|
||||
"""Mapping form lets workflow callers opt out of checkpoint persistence."""
|
||||
workflow = _build_simple_workflow()
|
||||
host = AgentFrameworkHost(
|
||||
target=workflow,
|
||||
channels=[_ChannelStub()],
|
||||
state_dir={"sessions": tmp_path / "s"},
|
||||
)
|
||||
try:
|
||||
assert host._checkpoint_location is None
|
||||
finally:
|
||||
_close_host_disk(host)
|
||||
|
||||
|
||||
def test_explicit_checkpoint_location_wins_over_state_dir(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""``checkpoint_location`` + ``state_dir`` → explicit param wins + warn."""
|
||||
workflow = _build_simple_workflow()
|
||||
explicit = tmp_path / "explicit-ck"
|
||||
with caplog.at_level("WARNING", logger="agent_framework.hosting"):
|
||||
host = AgentFrameworkHost(
|
||||
target=workflow,
|
||||
channels=[_ChannelStub()],
|
||||
checkpoint_location=explicit,
|
||||
state_dir=tmp_path,
|
||||
)
|
||||
try:
|
||||
assert host._checkpoint_location == explicit
|
||||
assert any(
|
||||
"state_dir['checkpoints']" in rec.message and "checkpoint_location" in rec.message for rec in caplog.records
|
||||
)
|
||||
finally:
|
||||
_close_host_disk(host)
|
||||
|
||||
|
||||
def test_state_dir_checkpoints_for_agent_target_silent_for_single_path(tmp_path: Path) -> None:
|
||||
"""Single-path state_dir + agent target → no checkpoint, no warning."""
|
||||
host = AgentFrameworkHost(
|
||||
target=_AgentStub(),
|
||||
channels=[_ChannelStub()],
|
||||
state_dir=tmp_path,
|
||||
)
|
||||
try:
|
||||
assert host._checkpoint_location is None
|
||||
assert not (tmp_path / "checkpoints").exists()
|
||||
finally:
|
||||
_close_host_disk(host)
|
||||
|
||||
|
||||
def test_state_dir_checkpoints_for_agent_target_warns_when_explicit(
|
||||
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Mapping form with ``checkpoints`` + agent target → warn."""
|
||||
with caplog.at_level("WARNING", logger="agent_framework.hosting"):
|
||||
host = AgentFrameworkHost(
|
||||
target=_AgentStub(),
|
||||
channels=[_ChannelStub()],
|
||||
state_dir={"checkpoints": tmp_path / "ck"},
|
||||
)
|
||||
try:
|
||||
assert host._checkpoint_location is None
|
||||
assert any(
|
||||
"state_dir['checkpoints']" in rec.message and "not a Workflow" in rec.message for rec in caplog.records
|
||||
)
|
||||
finally:
|
||||
_close_host_disk(host)
|
||||
|
||||
|
||||
def test_state_dir_checkpoints_conflicts_with_workflow_own_storage(tmp_path: Path) -> None:
|
||||
"""Derived checkpoint path triggers the same conflict guard as explicit."""
|
||||
from agent_framework import InMemoryCheckpointStorage, WorkflowBuilder
|
||||
|
||||
_UpperExecutor = importlib.import_module("hosting_workflow_fixtures")._UpperExecutor
|
||||
workflow = WorkflowBuilder(
|
||||
start_executor=_UpperExecutor(id="upper"),
|
||||
checkpoint_storage=InMemoryCheckpointStorage(),
|
||||
).build()
|
||||
with pytest.raises(RuntimeError, match="already has checkpoint storage"):
|
||||
AgentFrameworkHost(
|
||||
target=workflow,
|
||||
channels=[_ChannelStub()],
|
||||
state_dir=tmp_path,
|
||||
)
|
||||
@@ -1,316 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Tests for the per-request isolation contextvar surface in
|
||||
:mod:`agent_framework_hosting._isolation`.
|
||||
|
||||
The isolation keys are the ONLY seam Foundry-aware providers use to
|
||||
find partition keys, and the host's ASGI middleware lifts them off the
|
||||
two well-known headers on every inbound HTTP request. A regression
|
||||
that drops the lookup, mistypes a header name, or fails to reset the
|
||||
contextvar would silently misroute writes / leak per-request state
|
||||
across requests, with zero unit-test signal — so cover the surface
|
||||
fully here.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from agent_framework import AgentSession
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import JSONResponse
|
||||
from starlette.routing import BaseRoute, Route
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from agent_framework_hosting import (
|
||||
AgentFrameworkHost,
|
||||
Channel,
|
||||
ChannelContext,
|
||||
ChannelContribution,
|
||||
IsolationKeys,
|
||||
get_current_isolation_keys,
|
||||
reset_current_isolation_keys,
|
||||
set_current_isolation_keys,
|
||||
)
|
||||
from agent_framework_hosting._isolation import ( # pyright: ignore[reportPrivateUsage]
|
||||
ISOLATION_HEADER_CHAT,
|
||||
ISOLATION_HEADER_USER,
|
||||
current_isolation_keys,
|
||||
)
|
||||
|
||||
|
||||
class TestIsolationKeys:
|
||||
def test_defaults_to_none_pair(self) -> None:
|
||||
keys = IsolationKeys()
|
||||
assert keys.user_key is None
|
||||
assert keys.chat_key is None
|
||||
assert keys.is_empty is True
|
||||
|
||||
def test_partial_with_only_user_is_not_empty(self) -> None:
|
||||
keys = IsolationKeys(user_key="alice")
|
||||
assert keys.user_key == "alice"
|
||||
assert keys.chat_key is None
|
||||
assert keys.is_empty is False
|
||||
|
||||
def test_partial_with_only_chat_is_not_empty(self) -> None:
|
||||
keys = IsolationKeys(chat_key="general")
|
||||
assert keys.is_empty is False
|
||||
|
||||
def test_full_pair_is_not_empty(self) -> None:
|
||||
keys = IsolationKeys(user_key="alice", chat_key="general")
|
||||
assert keys.is_empty is False
|
||||
|
||||
|
||||
class TestContextVarHelpers:
|
||||
def test_default_is_none(self) -> None:
|
||||
# Each test gets a fresh contextvar value because pytest runs
|
||||
# tests in fresh contexts. ``get`` returns the default.
|
||||
assert get_current_isolation_keys() is None
|
||||
|
||||
def test_set_and_get_round_trip(self) -> None:
|
||||
token = set_current_isolation_keys(IsolationKeys(user_key="alice", chat_key="general"))
|
||||
try:
|
||||
current = get_current_isolation_keys()
|
||||
assert current is not None
|
||||
assert current.user_key == "alice"
|
||||
assert current.chat_key == "general"
|
||||
finally:
|
||||
reset_current_isolation_keys(token)
|
||||
# Reset restores prior value (None in the default context).
|
||||
assert get_current_isolation_keys() is None
|
||||
|
||||
def test_set_with_none_clears(self) -> None:
|
||||
outer = set_current_isolation_keys(IsolationKeys(user_key="alice"))
|
||||
try:
|
||||
inner = set_current_isolation_keys(None)
|
||||
try:
|
||||
assert get_current_isolation_keys() is None
|
||||
finally:
|
||||
reset_current_isolation_keys(inner)
|
||||
# Reset surfaces the outer value again.
|
||||
current = get_current_isolation_keys()
|
||||
assert current is not None
|
||||
assert current.user_key == "alice"
|
||||
finally:
|
||||
reset_current_isolation_keys(outer)
|
||||
|
||||
def test_module_level_contextvar_is_the_same_instance(self) -> None:
|
||||
"""Direct contextvar access (used by the ASGI middleware) and the
|
||||
public `get_current_isolation_keys()` helper read from the SAME
|
||||
underlying contextvar. A regression that introduced a second
|
||||
contextvar would silently break the middleware → provider hop."""
|
||||
token = current_isolation_keys.set(IsolationKeys(user_key="bob"))
|
||||
try:
|
||||
via_helper = get_current_isolation_keys()
|
||||
assert via_helper is not None
|
||||
assert via_helper.user_key == "bob"
|
||||
finally:
|
||||
current_isolation_keys.reset(token)
|
||||
|
||||
|
||||
class TestHeaderConstants:
|
||||
"""The two header names are part of the public contract — they
|
||||
match the ones the Foundry Hosted Agents runtime stamps on every
|
||||
inbound request. A typo here would silently misroute partition
|
||||
writes."""
|
||||
|
||||
def test_user_header_value(self) -> None:
|
||||
assert ISOLATION_HEADER_USER == "x-agent-user-isolation-key"
|
||||
|
||||
def test_chat_header_value(self) -> None:
|
||||
assert ISOLATION_HEADER_CHAT == "x-agent-chat-isolation-key"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# End-to-end: ASGI middleware lifts the headers into the contextvar.
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class _IsolationProbeChannel:
|
||||
"""A minimal Channel that exposes a single GET route which captures
|
||||
the contextvar value INSIDE the request and returns it as JSON.
|
||||
|
||||
Tests use this to exercise the full middleware → contextvar →
|
||||
handler hop end-to-end.
|
||||
"""
|
||||
|
||||
name = "probe"
|
||||
path = ""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.captured: list[IsolationKeys | None] = []
|
||||
|
||||
async def _handler(_request: Request) -> JSONResponse:
|
||||
keys = get_current_isolation_keys()
|
||||
self.captured.append(keys)
|
||||
payload: dict[str, str | bool | None]
|
||||
payload = (
|
||||
{"user": keys.user_key, "chat": keys.chat_key}
|
||||
if keys is not None
|
||||
else {"user": None, "chat": None, "_present": False}
|
||||
)
|
||||
return JSONResponse(payload)
|
||||
|
||||
self._routes: list[BaseRoute] = [Route("/probe", _handler)]
|
||||
|
||||
def contribute(self, context: ChannelContext) -> ChannelContribution:
|
||||
del context
|
||||
return ChannelContribution(routes=self._routes)
|
||||
|
||||
|
||||
def _make_host_with_probe() -> tuple[AgentFrameworkHost, _IsolationProbeChannel]:
|
||||
class _NoopAgent:
|
||||
id = "noop-agent"
|
||||
name: str | None = "Noop Agent"
|
||||
description: str | None = "Test noop agent"
|
||||
|
||||
def create_session(self, *, session_id: str | None = None) -> AgentSession:
|
||||
return AgentSession(session_id=session_id)
|
||||
|
||||
def get_session(self, service_session_id: str, *, session_id: str | None = None) -> AgentSession:
|
||||
return AgentSession(service_session_id=service_session_id, session_id=session_id)
|
||||
|
||||
def run(self, *_args: object, **_kwargs: object) -> Any: # pragma: no cover - never called
|
||||
raise RuntimeError("not invoked")
|
||||
|
||||
probe = _IsolationProbeChannel()
|
||||
assert isinstance(probe, Channel)
|
||||
host = AgentFrameworkHost(target=_NoopAgent(), channels=[probe])
|
||||
return host, probe
|
||||
|
||||
|
||||
class TestIsolationMiddlewareEndToEnd:
|
||||
def test_headers_ignored_outside_foundry_environment(self) -> None:
|
||||
host, probe = _make_host_with_probe()
|
||||
with TestClient(host.app) as client: # type: ignore[attr-defined]
|
||||
r = client.get(
|
||||
"/probe",
|
||||
headers={
|
||||
ISOLATION_HEADER_USER: "alice-uid",
|
||||
ISOLATION_HEADER_CHAT: "general-cid",
|
||||
},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"user": None, "chat": None, "_present": False}
|
||||
assert probe.captured == [None]
|
||||
|
||||
def test_both_headers_lifted_into_contextvar(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
|
||||
host, probe = _make_host_with_probe()
|
||||
with TestClient(host.app) as client: # type: ignore[attr-defined]
|
||||
r = client.get(
|
||||
"/probe",
|
||||
headers={
|
||||
ISOLATION_HEADER_USER: "alice-uid",
|
||||
ISOLATION_HEADER_CHAT: "general-cid",
|
||||
},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"user": "alice-uid", "chat": "general-cid"}
|
||||
assert len(probe.captured) == 1
|
||||
captured = probe.captured[0]
|
||||
assert captured is not None
|
||||
assert captured.user_key == "alice-uid"
|
||||
assert captured.chat_key == "general-cid"
|
||||
|
||||
def test_only_user_header_lifted(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""One-header-only branch: the middleware still binds (chat=None)."""
|
||||
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
|
||||
host, probe = _make_host_with_probe()
|
||||
with TestClient(host.app) as client: # type: ignore[attr-defined]
|
||||
r = client.get("/probe", headers={ISOLATION_HEADER_USER: "alice-uid"})
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"user": "alice-uid", "chat": None}
|
||||
|
||||
def test_only_chat_header_lifted(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
|
||||
host, probe = _make_host_with_probe()
|
||||
with TestClient(host.app) as client: # type: ignore[attr-defined]
|
||||
r = client.get("/probe", headers={ISOLATION_HEADER_CHAT: "general-cid"})
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"user": None, "chat": "general-cid"}
|
||||
|
||||
def test_no_headers_keeps_contextvar_none(self) -> None:
|
||||
"""Local-dev path: with neither header present the middleware is
|
||||
a no-op and the contextvar stays at its default ``None`` —
|
||||
providers see "no isolation" and route to the in-memory
|
||||
fallback rather than picking up stale per-request state."""
|
||||
host, probe = _make_host_with_probe()
|
||||
with TestClient(host.app) as client: # type: ignore[attr-defined]
|
||||
r = client.get("/probe")
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"user": None, "chat": None, "_present": False}
|
||||
assert probe.captured == [None]
|
||||
|
||||
def test_empty_header_value_treated_as_absent(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A header that's present but empty must not bind an empty key —
|
||||
``IsolationContext`` rejects empty strings on the read side."""
|
||||
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
|
||||
host, probe = _make_host_with_probe()
|
||||
with TestClient(host.app) as client: # type: ignore[attr-defined]
|
||||
r = client.get(
|
||||
"/probe",
|
||||
headers={
|
||||
ISOLATION_HEADER_USER: "",
|
||||
ISOLATION_HEADER_CHAT: "general-cid",
|
||||
},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
# Empty user header decodes to None; chat key stays bound.
|
||||
assert r.json() == {"user": None, "chat": "general-cid"}
|
||||
|
||||
def test_contextvar_resets_after_request(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""The middleware must call ``reset_current_isolation_keys`` in
|
||||
a ``finally`` so per-request state never leaks across requests
|
||||
or back into the calling thread's context."""
|
||||
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
|
||||
host, probe = _make_host_with_probe()
|
||||
with TestClient(host.app) as client: # type: ignore[attr-defined]
|
||||
r1 = client.get("/probe", headers={ISOLATION_HEADER_USER: "alice-uid"})
|
||||
assert r1.status_code == 200
|
||||
# Reading the contextvar OUTSIDE the request scope must see
|
||||
# the default — not the value the prior request bound.
|
||||
assert get_current_isolation_keys() is None
|
||||
# And a follow-up request without headers gets a clean
|
||||
# ``None`` rather than inheriting alice-uid.
|
||||
r2 = client.get("/probe")
|
||||
assert r2.json() == {"user": None, "chat": None, "_present": False}
|
||||
|
||||
def test_concurrent_requests_get_isolated_contextvars(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Different requests run in different async contexts; binding
|
||||
from request A must NOT leak into a concurrent request B."""
|
||||
monkeypatch.setenv("FOUNDRY_HOSTING_ENVIRONMENT", "1")
|
||||
host, probe = _make_host_with_probe()
|
||||
|
||||
async def _drive() -> None:
|
||||
# Run two requests in parallel asyncio tasks against the
|
||||
# same TestClient and assert their captures don't bleed
|
||||
# into each other.
|
||||
async def _hit(user_key: str) -> dict[str, str | None]:
|
||||
with TestClient(host.app) as client: # type: ignore[attr-defined]
|
||||
r = client.get("/probe", headers={ISOLATION_HEADER_USER: user_key})
|
||||
return r.json() # type: ignore[no-any-return]
|
||||
|
||||
r_alice, r_bob = await asyncio.gather(_hit("alice-uid"), _hit("bob-uid"))
|
||||
assert r_alice == {"user": "alice-uid", "chat": None}
|
||||
assert r_bob == {"user": "bob-uid", "chat": None}
|
||||
|
||||
asyncio.run(_drive())
|
||||
|
||||
|
||||
class TestNonHttpScopesPassThrough:
|
||||
"""The middleware intentionally only inspects ``http`` scopes;
|
||||
lifespan / websocket scopes are forwarded untouched. A regression
|
||||
that touched lifespan scopes here would crash boot."""
|
||||
|
||||
async def test_lifespan_scope_does_not_consult_headers(self) -> None:
|
||||
# The TestClient context manager exercises the lifespan scope
|
||||
# implicitly; if the middleware tried to decode headers on a
|
||||
# non-http scope this would raise. Exercise it without binding
|
||||
# any contextvar work.
|
||||
host, _probe = _make_host_with_probe()
|
||||
with TestClient(host.app): # type: ignore[attr-defined]
|
||||
# Just enter / exit; no requests.
|
||||
pass
|
||||
@@ -1,50 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Tests for the channel-neutral envelope types in :mod:`agent_framework_hosting._types`."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from agent_framework_hosting import (
|
||||
ChannelIdentity,
|
||||
ChannelRequest,
|
||||
ChannelSession,
|
||||
)
|
||||
|
||||
|
||||
class TestChannelRequest:
|
||||
def test_required_fields_only(self) -> None:
|
||||
req = ChannelRequest(channel="responses", operation="message.create", input="hi")
|
||||
assert req.channel == "responses"
|
||||
assert req.operation == "message.create"
|
||||
assert req.input == "hi"
|
||||
assert req.session is None
|
||||
assert req.options is None
|
||||
assert req.session_mode == "auto"
|
||||
assert req.metadata == {}
|
||||
assert req.attributes == {}
|
||||
assert req.stream is False
|
||||
assert req.identity is None
|
||||
|
||||
def test_with_session_and_identity(self) -> None:
|
||||
req = ChannelRequest(
|
||||
channel="telegram",
|
||||
operation="message.create",
|
||||
input="hi",
|
||||
session=ChannelSession(isolation_key="user:42"),
|
||||
identity=ChannelIdentity(channel="telegram", native_id="42"),
|
||||
)
|
||||
assert req.session is not None
|
||||
assert req.session.isolation_key == "user:42"
|
||||
assert req.identity is not None
|
||||
assert req.identity.channel == "telegram"
|
||||
assert req.identity.native_id == "42"
|
||||
|
||||
|
||||
class TestChannelIdentity:
|
||||
def test_attributes_default_empty_mapping(self) -> None:
|
||||
ident = ChannelIdentity(channel="teams", native_id="abc")
|
||||
assert dict(ident.attributes) == {}
|
||||
|
||||
def test_attributes_passthrough(self) -> None:
|
||||
ident = ChannelIdentity(channel="teams", native_id="abc", attributes={"role": "user"})
|
||||
assert dict(ident.attributes) == {"role": "user"}
|
||||
@@ -10,7 +10,6 @@ available in CI / dev sandboxes).
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
@@ -25,27 +24,11 @@ from agent_framework_tools.shell._docker import (
|
||||
build_run_argv,
|
||||
)
|
||||
|
||||
|
||||
def _docker_image_available(image: str) -> bool:
|
||||
if not is_docker_available():
|
||||
return False
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["docker", "image", "inspect", image],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
timeout=5.0,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return False
|
||||
return result.returncode == 0
|
||||
|
||||
|
||||
# Integration tests use Linux container images (alpine) that don't run
|
||||
# under Docker Desktop's default Windows-container mode.
|
||||
_skip_if_no_linux_docker = pytest.mark.skipif(
|
||||
not _docker_image_available("alpine:3") or sys.platform == "win32",
|
||||
reason="docker daemon unavailable, alpine:3 image missing, or running Windows containers",
|
||||
not is_docker_available() or sys.platform == "win32",
|
||||
reason="docker daemon unavailable or running Windows containers",
|
||||
)
|
||||
|
||||
# --------------------------------------------------------------------- argv builders
|
||||
|
||||
@@ -91,8 +91,6 @@ agent-framework-foundry-hosting = { workspace = true }
|
||||
agent-framework-foundry-local = { workspace = true }
|
||||
agent-framework-gemini = { workspace = true }
|
||||
agent-framework-github-copilot = { workspace = true }
|
||||
agent-framework-hosting = { workspace = true }
|
||||
agent-framework-hosting-responses = { workspace = true }
|
||||
agent-framework-hyperlight = { workspace = true }
|
||||
agent-framework-lab = { workspace = true }
|
||||
agent-framework-mem0 = { workspace = true }
|
||||
|
||||
@@ -12,10 +12,6 @@ import asyncio
|
||||
from agent_framework import Agent, workflow
|
||||
from agent_framework.foundry import FoundryChatClient
|
||||
from azure.identity import AzureCliCredential
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load environment variables from .env file (e.g., FOUNDRY_PROJECT_ENDPOINT, FOUNDRY_MODEL)
|
||||
load_dotenv()
|
||||
|
||||
# <create_agents>
|
||||
client = FoundryChatClient(credential=AzureCliCredential())
|
||||
|
||||
@@ -24,9 +24,8 @@ async def main() -> None:
|
||||
Creates a task for the chat request, waits briefly, then cancels it to show proper cleanup.
|
||||
|
||||
Configuration:
|
||||
- FOUNDRY_PROJECT_ENDPOINT: Azure AI Foundry project endpoint URL
|
||||
- FOUNDRY_MODEL: Model deployment name (e.g. gpt-4o)
|
||||
- Authentication: Run `az login` to authenticate via AzureCliCredential
|
||||
- OpenAI model ID: Use "model" parameter or "OPENAI_MODEL" environment variable
|
||||
- OpenAI API key: Use "api_key" parameter or "OPENAI_API_KEY" environment variable
|
||||
"""
|
||||
client = FoundryChatClient(credential=AzureCliCredential())
|
||||
|
||||
|
||||
@@ -1,42 +0,0 @@
|
||||
# Multi-channel hosting samples
|
||||
|
||||
End-to-end samples for serving an `agent-framework` agent (or workflow)
|
||||
through one or more **channels** with `agent-framework-hosting`.
|
||||
|
||||
The general hosting plumbing lives in
|
||||
[`agent-framework-hosting`](../../../packages/hosting); each channel is
|
||||
its own package. This first sample set includes
|
||||
`agent-framework-hosting-responses`.
|
||||
|
||||
| Sample | What it shows | Packaging |
|
||||
|---|---|---|
|
||||
| [`local_responses/`](./local_responses) | The minimal shape: one agent + one `@tool` + `ResponsesChannel` + a single `run_hook` that strips caller-supplied options and forces a `reasoning` preset. | **Local only.** Start here to learn the run-hook seam. |
|
||||
| [`local_responses_workflow/`](./local_responses_workflow) | A 4-step `Workflow` (typed `SloganBrief` intake → writer → legal → formatter) hosted behind the Responses channel via a `run_hook` that parses inbound text/JSON into the workflow's typed input. The host writes per-conversation checkpoints via `checkpoint_location=…`. Demonstrates workflow targets + structured input adaptation + resume-across-turns. Includes a `call_server.rest` file with REST examples. | **Local only.** |
|
||||
|
||||
Each sample is fully self-contained — its own `pyproject.toml`, `uv.lock`,
|
||||
server `app.py`, calling script(s), and `storage/` directory. Every
|
||||
sample uses `[tool.uv.sources]` to wire its `agent-framework-hosting*`
|
||||
dependencies to the
|
||||
[`main`](https://github.com/microsoft/agent-framework/tree/main)
|
||||
branch of the upstream repo via git refs, so they install cleanly outside
|
||||
the monorepo while the hosting packages are still pre-PyPI. Once those
|
||||
packages publish, drop the `[tool.uv.sources]` block and let the
|
||||
declared deps resolve from PyPI.
|
||||
|
||||
## Relationship to `../foundry-hosted-agents/`
|
||||
|
||||
The sibling [`../foundry-hosted-agents/`](../foundry-hosted-agents) directory
|
||||
contains samples for the **`agent-framework-hosted`** stack — agents
|
||||
that run **inside** the Foundry Hosted Agents platform using its
|
||||
built-in protocol surface (Responses, Invocations, conversation store,
|
||||
isolation, identity), with **no `agent-framework-hosting` package
|
||||
involved**.
|
||||
|
||||
| Aspect | `af-hosting/` (this directory) | `foundry-hosted-agents/` |
|
||||
|---|---|---|
|
||||
| Server stack | `agent-framework-hosting` + `agent-framework-hosting-responses` | `agent-framework-hosted` only — the Foundry Hosted Agents runtime owns the HTTP surface |
|
||||
| Channels | Responses only in this initial sample set | The platform exposes Responses + Invocations |
|
||||
| Run target | Local Hypercorn (`local_responses/`, `local_responses_workflow/`) | Hosted Agents *or* local container; targets the Hosted Agents platform contract |
|
||||
| When to pick this | You want to learn the host/channel seams locally or need custom hosting middleware | You want zero hosting boilerplate, leveraging the Foundry-managed surface |
|
||||
|
||||
The table above summarizes the cross-sample story.
|
||||
@@ -1,54 +0,0 @@
|
||||
# local_responses — Responses-only with a settings-altering hook
|
||||
|
||||
The smallest end-to-end `agent-framework-hosting` shape: one Foundry
|
||||
agent with a `@tool`, one `ResponsesChannel`, one `run_hook`. Useful as
|
||||
the entry-point sample for understanding the **channel run-hook** seam
|
||||
without any multi-channel or identity-link concerns.
|
||||
|
||||
What the run hook demonstrates:
|
||||
|
||||
- **Strips** caller-supplied `model` / `temperature` / `store` so the
|
||||
host owns the backing deployment and persistence settings.
|
||||
- **Forces** a `reasoning` preset (`effort=medium`, `summary=auto`) on
|
||||
every turn — caller-side overrides are ignored.
|
||||
|
||||
`app:app` is a module-level Starlette ASGI app; recommended local launch
|
||||
is Hypercorn.
|
||||
|
||||
## Run
|
||||
|
||||
```bash
|
||||
export FOUNDRY_PROJECT_ENDPOINT=https://<your-project>.services.ai.azure.com
|
||||
export FOUNDRY_MODEL=gpt-5-nano
|
||||
az login
|
||||
|
||||
uv sync
|
||||
uv run hypercorn app:app --bind 0.0.0.0:8000
|
||||
```
|
||||
|
||||
Single-process for quick iteration:
|
||||
|
||||
```bash
|
||||
uv run python app.py
|
||||
```
|
||||
|
||||
## Call locally
|
||||
|
||||
```bash
|
||||
uv sync --group dev
|
||||
|
||||
# Plain OpenAI SDK call:
|
||||
uv run python call_server.py
|
||||
|
||||
# The client intentionally omits `model`; the host chooses the backing
|
||||
# deployment from FOUNDRY_MODEL.
|
||||
|
||||
# The script then sends a second turn, "And what about Amsterdam?",
|
||||
# using the first `response.id` as `previous_response_id`.
|
||||
|
||||
# Same two-turn interaction through an Agent Framework Agent backed by
|
||||
# OpenAIChatClient, with streaming enabled:
|
||||
uv run python call_server_af.py
|
||||
```
|
||||
|
||||
> This sample is **local-only** — no Dockerfile, no Foundry packaging.
|
||||
@@ -1,126 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Minimal Responses-only hosting sample.
|
||||
|
||||
Single agent with one ``@tool`` (``lookup_weather``), single channel
|
||||
(``ResponsesChannel``), one ``run_hook`` that demonstrates the
|
||||
settings-mutation seam over caller-supplied options.
|
||||
|
||||
What the hook does
|
||||
------------------
|
||||
On every Responses request the hook receives the ``ChannelRequest`` that
|
||||
the channel built from the inbound HTTP body. It:
|
||||
|
||||
- strips ``model`` (the host owns the backing deployment), ``store``
|
||||
(this agent owns persistence), and ``temperature`` (the configured
|
||||
model may not honor it),
|
||||
- forces a ``reasoning`` effort + summary preset so the deployed surface
|
||||
is consistent regardless of what the caller sent.
|
||||
|
||||
The hook is the documented escape hatch over the uniform
|
||||
``ChannelRequest`` envelope.
|
||||
|
||||
Run
|
||||
---
|
||||
``app`` is a module-level Starlette ASGI app. Recommended local launch::
|
||||
|
||||
uv sync
|
||||
az login
|
||||
export FOUNDRY_PROJECT_ENDPOINT=https://<your-project>.services.ai.azure.com
|
||||
export FOUNDRY_MODEL=gpt-5-nano
|
||||
uv run hypercorn app:app --bind 0.0.0.0:8000
|
||||
|
||||
Or use the ``__main__`` block (single-process Hypercorn) for quick
|
||||
iteration::
|
||||
|
||||
uv run python app.py
|
||||
|
||||
Then call it::
|
||||
|
||||
uv run python call_server.py "What is the weather in Tokyo?"
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import replace
|
||||
from pathlib import Path
|
||||
from typing import Annotated
|
||||
|
||||
from agent_framework import Agent, FileHistoryProvider, tool
|
||||
from agent_framework_foundry import FoundryChatClient
|
||||
from agent_framework_hosting import AgentFrameworkHost, ChannelRequest
|
||||
from agent_framework_hosting_responses import ResponsesChannel
|
||||
from azure.identity.aio import DefaultAzureCredential
|
||||
|
||||
SESSIONS_DIR = Path(__file__).resolve().parent / "storage" / "sessions"
|
||||
SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
@tool(approval_mode="never_require")
|
||||
def lookup_weather(
|
||||
location: Annotated[str, "The city to look up weather for."],
|
||||
) -> str:
|
||||
"""Return a deterministic weather report for a city."""
|
||||
high_temp = 5 + (sum(location.encode("utf-8")) % 21)
|
||||
reports = {
|
||||
"Seattle": f"Seattle is rainy with a high of {high_temp}°C.",
|
||||
"Amsterdam": f"Amsterdam is cloudy with a high of {high_temp}°C.",
|
||||
"Tokyo": f"Tokyo is clear with a high of {high_temp}°C.",
|
||||
}
|
||||
return reports.get(location, f"{location} is sunny with a high of {high_temp}°C.")
|
||||
|
||||
|
||||
# the run hook defines what you want to allow the user to passthrough when they call your host
|
||||
# since the responses clients can call with all of the responses options,
|
||||
# you can decide with this run_hook which of those: are rejected
|
||||
# which are passed through, which are altered, which are added.
|
||||
# In this sample below, we are removing, model, temperature and store if set
|
||||
# and we add reasoning, but note that this could also be set on the Agent itself
|
||||
# the difference is that this option is specific to the Responses channel
|
||||
# so if you want to differentiate between options over channels
|
||||
# you would set the option in the run_hook, if it needs to be the same (like store)
|
||||
# you would set it in the agent.
|
||||
def run_hook(request: ChannelRequest, **_: object) -> ChannelRequest:
|
||||
"""Strip caller-supplied options the host should own and force a
|
||||
reasoning preset."""
|
||||
options = dict(request.options or {})
|
||||
|
||||
# The host owns the backing deployment; the agent's default_options
|
||||
# own ``store``; the model may not honor ``temperature``. Strip them
|
||||
# so the caller can't override.
|
||||
options.pop("model", None)
|
||||
options.pop("temperature", None)
|
||||
options.pop("store", None)
|
||||
|
||||
# Force a consistent reasoning preset on every turn.
|
||||
options["reasoning"] = {"effort": "medium", "summary": "auto"}
|
||||
|
||||
return replace(request, options=options or None)
|
||||
|
||||
|
||||
def build_host() -> AgentFrameworkHost:
|
||||
# Here we define how our agent should run, with tools, options, etc:
|
||||
agent = Agent(
|
||||
client=FoundryChatClient(credential=DefaultAzureCredential()),
|
||||
name="WeatherAgent",
|
||||
instructions=(
|
||||
"You are a friendly weather assistant. Use the lookup_weather tool "
|
||||
"for any weather question and answer in one short sentence."
|
||||
),
|
||||
tools=[lookup_weather],
|
||||
context_providers=[FileHistoryProvider(SESSIONS_DIR)],
|
||||
default_options={"store": False},
|
||||
)
|
||||
return AgentFrameworkHost(
|
||||
target=agent,
|
||||
channels=[ResponsesChannel(run_hook=run_hook)],
|
||||
debug=True,
|
||||
)
|
||||
|
||||
|
||||
app = build_host().app
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
build_host().serve(host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))
|
||||
@@ -1,51 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Local client for the local_responses sample.
|
||||
|
||||
Posts to ``/responses`` using the standard ``openai`` SDK.
|
||||
|
||||
Pass ``--previous-response-id <id>`` to continue a conversation by its
|
||||
``response.id`` (returned in the prior response).
|
||||
|
||||
Start the server first (in another shell)::
|
||||
|
||||
uv run python app.py
|
||||
|
||||
Then::
|
||||
|
||||
uv run python call_server.py
|
||||
|
||||
The script sends a follow-up turn ("And what about Amsterdam?") using the
|
||||
first response's ``response.id`` as ``previous_response_id``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from openai import OpenAI
|
||||
|
||||
BASE_URL = "http://127.0.0.1:8000"
|
||||
PROMPT = "What is the weather in Tokyo?"
|
||||
FOLLOW_UP_PROMPT = "And what about Amsterdam?"
|
||||
|
||||
|
||||
def main() -> None:
|
||||
client = OpenAI(base_url=BASE_URL, api_key="not-needed")
|
||||
response = client.responses.create(
|
||||
input=PROMPT,
|
||||
)
|
||||
print(f"User: {PROMPT}")
|
||||
print(f"Agent: {response.output_text}")
|
||||
print(f"Response ID: {response.id}")
|
||||
|
||||
follow_up = client.responses.create(
|
||||
input=FOLLOW_UP_PROMPT,
|
||||
previous_response_id=response.id,
|
||||
)
|
||||
print()
|
||||
print(f"User: {FOLLOW_UP_PROMPT}")
|
||||
print(f"Agent: {follow_up.output_text}")
|
||||
print(f"Response ID: {follow_up.id}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,60 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Agent Framework agent client for the local_responses sample.
|
||||
|
||||
Creates a local :class:`agent_framework.Agent` backed by
|
||||
:class:`agent_framework.openai.OpenAIChatClient`, points that client at the
|
||||
hosted ``/responses`` endpoint, and streams both turns:
|
||||
|
||||
1. ``What is the weather in Tokyo?``
|
||||
2. ``And what about Amsterdam?``
|
||||
|
||||
Both turns use the same :class:`agent_framework.AgentSession`; the first
|
||||
turn binds the hosted response id to the session, and the second turn
|
||||
continues through that session.
|
||||
|
||||
Start the server first (in another shell)::
|
||||
|
||||
uv run python app.py
|
||||
|
||||
Then::
|
||||
|
||||
uv run python call_server_af.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
from agent_framework import Agent
|
||||
from agent_framework.openai import OpenAIChatClient
|
||||
|
||||
BASE_URL = "http://127.0.0.1:8000"
|
||||
PROMPTS = [
|
||||
"What is the weather in Tokyo?",
|
||||
"And what about Amsterdam?",
|
||||
]
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
agent = Agent(
|
||||
client=OpenAIChatClient(base_url=BASE_URL, api_key="not-needed"),
|
||||
name="HostedWeatherClient",
|
||||
)
|
||||
session = agent.create_session()
|
||||
|
||||
for prompt in PROMPTS:
|
||||
print(f"User: {prompt}")
|
||||
stream = agent.run(prompt, stream=True, session=session)
|
||||
print("Agent: ", end="", flush=True)
|
||||
async for update in stream:
|
||||
if update.text:
|
||||
print(update.text, end="", flush=True)
|
||||
|
||||
response = await stream.get_final_response()
|
||||
print("\n")
|
||||
print(f"Response ID: {response.response_id}\n")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,27 +0,0 @@
|
||||
[project]
|
||||
name = "agent-framework-hosting-sample-local-responses"
|
||||
version = "0.0.1"
|
||||
description = "Minimal Responses-only local hosting sample with a settings-altering run hook."
|
||||
requires-python = ">=3.10"
|
||||
dependencies = [
|
||||
"agent-framework-foundry",
|
||||
"agent-framework-hosting",
|
||||
"agent-framework-hosting-responses",
|
||||
"azure-identity",
|
||||
"aiohttp>=3.13.5",
|
||||
"hypercorn>=0.17",
|
||||
]
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"agent-framework-openai",
|
||||
"openai>=1.99",
|
||||
]
|
||||
|
||||
[tool.uv]
|
||||
package = false
|
||||
|
||||
[tool.uv.sources]
|
||||
agent-framework-hosting = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting" }
|
||||
agent-framework-hosting-responses = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting-responses" }
|
||||
agent-framework-openai = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/openai" }
|
||||
@@ -1,83 +0,0 @@
|
||||
# local_responses_workflow — workflow target with run-hook prep + checkpoints
|
||||
|
||||
A `Workflow` (writer → legal reviewer → formatter) hosted
|
||||
behind the **Responses API**, with the host configured to
|
||||
**persist per-conversation checkpoints**. Mirrors
|
||||
[`../../foundry-hosted-agents/responses/05_workflows/`](../../foundry-hosted-agents/responses/05_workflows/)
|
||||
but uses the `agent-framework-hosting` stack instead of the
|
||||
Foundry-Hosted-Agents runtime. The `run_hook` prepares the writer prompt
|
||||
before the workflow starts.
|
||||
|
||||
## What's interesting
|
||||
|
||||
- `AgentFrameworkHost(target=workflow, …)` — the host detects a
|
||||
`Workflow` target and dispatches to `workflow.run(...)` (no
|
||||
`Agent.create_session(...)`).
|
||||
- `ResponsesChannel` is mounted at `/responses` with a `prepare_writer_prompt`
|
||||
run hook that **adapts the channel-native input into the workflow start
|
||||
executor's input**. Responses delivers a `list[Message]`; the hook normalises
|
||||
it to text and prepares the prompt the writer agent receives.
|
||||
- The hook parses the inbound text as JSON
|
||||
(`{"topic": ..., "style": ..., "audience": ...}`); if parsing fails
|
||||
it uses the whole text as `topic` with defaults.
|
||||
- The workflow starts directly at the writer `AgentExecutor`; no extra intake
|
||||
executor is needed because the hook performs the one preparation step.
|
||||
- `checkpoint_location=storage/checkpoints/` — the host scopes a
|
||||
`FileCheckpointStorage` per conversation (Responses keys it on
|
||||
`previous_response_id` / `conversation_id`) and **restores from the
|
||||
latest checkpoint at the start of every turn** before applying the new
|
||||
input. Without an isolation key the host skips checkpointing for that request.
|
||||
- No `HistoryProvider` — the workflow owns its own state via the
|
||||
checkpoint store.
|
||||
|
||||
## Run
|
||||
|
||||
```bash
|
||||
export FOUNDRY_PROJECT_ENDPOINT=https://<your-project>.services.ai.azure.com
|
||||
export FOUNDRY_MODEL=gpt-5-nano
|
||||
az login
|
||||
|
||||
uv sync
|
||||
uv run hypercorn app:app --bind 0.0.0.0:8000
|
||||
```
|
||||
|
||||
Single-process for quick iteration:
|
||||
|
||||
```bash
|
||||
uv run python app.py
|
||||
```
|
||||
|
||||
## Call locally
|
||||
|
||||
Two clients are provided next to `app.py`:
|
||||
|
||||
- **`call_server.py`** — Python client using the OpenAI SDK (Responses
|
||||
API only).
|
||||
- **`call_server.rest`** — raw REST examples for the Responses endpoint
|
||||
(open in VS Code with the REST Client extension or any compatible HTTP-file
|
||||
runner).
|
||||
|
||||
```bash
|
||||
uv sync --group dev
|
||||
|
||||
# Structured brief via the OpenAI SDK (Responses API):
|
||||
uv run python call_server.py \
|
||||
'{"topic": "electric SUV", "style": "playful", "audience": "young families"}'
|
||||
|
||||
# The client intentionally omits `model`; the host chooses the backing
|
||||
# deployment from FOUNDRY_MODEL.
|
||||
|
||||
# Plain topic (style/audience default to "modern" / "general"):
|
||||
uv run python call_server.py "electric SUV"
|
||||
|
||||
# Continue an existing conversation by its `response.id`:
|
||||
uv run python call_server.py --previous-response-id <response-id> \
|
||||
'{"topic": "electric SUV", "style": "retro", "audience": "boomers"}'
|
||||
```
|
||||
|
||||
After a few turns, inspect `storage/checkpoints/<isolation_key>/` —
|
||||
each conversation has its own subdirectory of checkpoint files written
|
||||
by the host.
|
||||
|
||||
> This sample is **local-only** — no Dockerfile, no Foundry packaging.
|
||||
> A Foundry-Hosted-Agents-compatible packaging sample will be added separately.
|
||||
@@ -1,182 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Hosted workflow sample with run-hook input prep + checkpoint location.
|
||||
|
||||
Same three-agent slogan workflow as
|
||||
``../../foundry-hosted-agents/responses/05_workflows/main.py`` (writer →
|
||||
legal reviewer → formatter), driven through the ``agent-framework-hosting``
|
||||
stack instead of the Foundry-Hosted-Agents runtime.
|
||||
|
||||
Workflow shape
|
||||
--------------
|
||||
``writer`` → ``legal_reviewer`` → ``formatter``. A single run hook parses
|
||||
the Responses input and prepares the prompt the writer agent receives.
|
||||
|
||||
What this sample shows
|
||||
----------------------
|
||||
- A :class:`~agent_framework.Workflow` is a valid hosting target — the
|
||||
host detects it and dispatches to ``workflow.run(...)`` instead of
|
||||
``agent.run(...)``.
|
||||
- ``ResponsesChannel(run_hook=...)`` is the seam for **adapting the
|
||||
channel-native input into the workflow start executor's input**.
|
||||
The hook here parses the inbound text as JSON
|
||||
(``{"topic": ..., "style": ..., "audience": ...}``) — if parsing
|
||||
fails it falls back to using the whole text as ``topic`` with
|
||||
defaults — and replaces ``ChannelRequest.input`` with the prepared
|
||||
writer prompt.
|
||||
- ``AgentFrameworkHost(checkpoint_location=...)`` enables
|
||||
per-conversation workflow checkpointing. The host scopes the
|
||||
checkpoint storage by ``ChannelRequest.session.isolation_key``
|
||||
(Responses uses ``previous_response_id`` / ``conversation_id`` as the
|
||||
isolation key), and restores from the latest checkpoint before each
|
||||
new turn — so a multi-turn workflow can resume across requests.
|
||||
- No ``HistoryProvider`` is configured: the workflow owns its own state
|
||||
via the checkpoint store; the agent-history seam is for plain
|
||||
``SupportsAgentRun`` agents.
|
||||
|
||||
Run
|
||||
---
|
||||
``app`` is a module-level Starlette ASGI app::
|
||||
|
||||
uv sync
|
||||
az login
|
||||
export FOUNDRY_PROJECT_ENDPOINT=https://<your-project>.services.ai.azure.com
|
||||
export FOUNDRY_MODEL=gpt-5-nano
|
||||
uv run hypercorn app:app --bind 0.0.0.0:8000
|
||||
|
||||
Or for quick iteration::
|
||||
|
||||
uv run python app.py
|
||||
|
||||
Then call it with a structured brief::
|
||||
|
||||
uv run python call_server.py \\
|
||||
'{"topic": "electric SUV", "style": "playful", "audience": "young families"}'
|
||||
|
||||
Or with just a topic — the hook fills in defaults::
|
||||
|
||||
uv run python call_server.py "Create a slogan for an electric SUV."
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from dataclasses import replace
|
||||
from pathlib import Path
|
||||
|
||||
from agent_framework import (
|
||||
Agent,
|
||||
AgentExecutor,
|
||||
Message,
|
||||
WorkflowBuilder,
|
||||
)
|
||||
from agent_framework_foundry import FoundryChatClient
|
||||
from agent_framework_hosting import AgentFrameworkHost, ChannelRequest
|
||||
from agent_framework_hosting_responses import ResponsesChannel
|
||||
from azure.identity.aio import DefaultAzureCredential
|
||||
|
||||
CHECKPOINTS_DIR = Path(__file__).resolve().parent / "storage" / "checkpoints"
|
||||
CHECKPOINTS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
def prepare_writer_prompt(request: ChannelRequest, **_: object) -> ChannelRequest:
|
||||
"""Prepare the workflow's initial writer prompt from Responses input.
|
||||
|
||||
The channel hands the host either a ``str`` (rare on the Responses
|
||||
surface) or a list of :class:`Message`. This hook collapses that
|
||||
input to text, accepts either JSON or plain text, and replaces the
|
||||
request input with a plain prompt for the writer executor.
|
||||
"""
|
||||
|
||||
def extract_text(value: object) -> str:
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
if isinstance(value, Message):
|
||||
return value.text
|
||||
if isinstance(value, list):
|
||||
return "\n".join(extract_text(item) for item in value)
|
||||
return ""
|
||||
|
||||
text = extract_text(request.input).strip()
|
||||
topic = text or "a generic product"
|
||||
style = "modern"
|
||||
audience = "general"
|
||||
if topic.startswith("{"):
|
||||
try:
|
||||
data = json.loads(topic)
|
||||
except json.JSONDecodeError:
|
||||
data = None
|
||||
if isinstance(data, dict) and "topic" in data:
|
||||
topic = str(data["topic"])
|
||||
style = str(data.get("style", style))
|
||||
audience = str(data.get("audience", audience))
|
||||
|
||||
prompt = (
|
||||
f"Topic: {topic}\n"
|
||||
f"Style: {style}\n"
|
||||
f"Audience: {audience}\n\n"
|
||||
"Write a single short slogan that fits the topic, style, and audience."
|
||||
)
|
||||
return replace(request, input=prompt)
|
||||
|
||||
|
||||
def build_host() -> AgentFrameworkHost:
|
||||
client = FoundryChatClient(credential=DefaultAzureCredential())
|
||||
|
||||
writer = Agent(
|
||||
client=client,
|
||||
name="writer",
|
||||
instructions=("You are an excellent slogan writer. You create new slogans based on the given topic."),
|
||||
)
|
||||
legal = Agent(
|
||||
client=client,
|
||||
name="legal_reviewer",
|
||||
instructions=(
|
||||
"You are an excellent legal reviewer. "
|
||||
"Make necessary corrections to the slogan so that it is legally compliant."
|
||||
),
|
||||
)
|
||||
formatter = Agent(
|
||||
client=client,
|
||||
name="formatter",
|
||||
instructions=(
|
||||
"You are an excellent content formatter. "
|
||||
"You take the slogan and format it in a cool retro style when printing to a terminal."
|
||||
),
|
||||
)
|
||||
|
||||
# ``context_mode="last_agent"`` ensures each agent only sees the
|
||||
# previous executor's output — matching the Foundry sample.
|
||||
writer_ex = AgentExecutor(writer, context_mode="last_agent")
|
||||
legal_ex = AgentExecutor(legal, context_mode="last_agent")
|
||||
format_ex = AgentExecutor(formatter, context_mode="last_agent")
|
||||
|
||||
workflow = (
|
||||
WorkflowBuilder(
|
||||
start_executor=writer_ex,
|
||||
output_executors=[format_ex],
|
||||
)
|
||||
.add_edge(writer_ex, legal_ex)
|
||||
.add_edge(legal_ex, format_ex)
|
||||
.build()
|
||||
)
|
||||
|
||||
return AgentFrameworkHost(
|
||||
target=workflow,
|
||||
channels=[
|
||||
ResponsesChannel(run_hook=prepare_writer_prompt),
|
||||
],
|
||||
# The host writes a per-conversation FileCheckpointStorage rooted
|
||||
# at ``CHECKPOINTS_DIR / <isolation_key>`` and restores from the
|
||||
# latest checkpoint at the start of every turn.
|
||||
checkpoint_location=CHECKPOINTS_DIR,
|
||||
debug=True,
|
||||
)
|
||||
|
||||
|
||||
app = build_host().app
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
build_host().serve(host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))
|
||||
@@ -1,53 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Local client for the local_responses_workflow sample.
|
||||
|
||||
The server expects a structured slogan brief. You can either pass a
|
||||
JSON object or a plain topic string (the server's run hook fills the
|
||||
other fields with defaults).
|
||||
|
||||
Pass ``--previous-response-id <id>`` to continue a conversation by its
|
||||
``response.id`` — the host uses that as the workflow checkpoint scope
|
||||
key, so the workflow resumes from where it left off.
|
||||
|
||||
Start the server first (in another shell)::
|
||||
|
||||
uv run python app.py
|
||||
|
||||
Then::
|
||||
|
||||
uv run python call_server.py \\
|
||||
'{"topic": "electric SUV", "style": "playful", "audience": "young families"}'
|
||||
|
||||
uv run python call_server.py "electric SUV" # uses default style/audience
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
from openai import OpenAI
|
||||
|
||||
BASE_URL = "http://127.0.0.1:8000"
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = sys.argv[1:]
|
||||
previous_response_id: str | None = None
|
||||
if len(args) >= 2 and args[0] == "--previous-response-id":
|
||||
previous_response_id = args[1]
|
||||
args = args[2:]
|
||||
print(f"Resuming response: {previous_response_id}")
|
||||
prompt = " ".join(args) or '{"topic": "electric SUV", "style": "playful", "audience": "young families"}'
|
||||
client = OpenAI(base_url=BASE_URL, api_key="not-needed")
|
||||
response = client.responses.create(
|
||||
input=prompt,
|
||||
previous_response_id=previous_response_id,
|
||||
)
|
||||
print(f"User: {prompt}")
|
||||
print(f"Agent: {response.output_text}")
|
||||
print(f"response.id: {response.id}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,48 +0,0 @@
|
||||
# local_responses_workflow — REST examples
|
||||
#
|
||||
# Use with the VS Code "REST Client" extension (humao.rest-client) or
|
||||
# JetBrains HTTP Client. Each `###` block is one request.
|
||||
#
|
||||
# Start the server in another shell first:
|
||||
# uv run python app.py
|
||||
|
||||
@host = http://127.0.0.1:8000
|
||||
|
||||
###
|
||||
# 1. Responses API — structured brief
|
||||
POST {{host}}/responses
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"model": "agent",
|
||||
"input": "{\"topic\": \"electric SUV\", \"style\": \"playful\", \"audience\": \"young families\"}"
|
||||
}
|
||||
|
||||
###
|
||||
# 2. Responses API — plain topic, defaults applied by the run hook
|
||||
POST {{host}}/responses
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"model": "agent",
|
||||
"input": "vintage espresso machine"
|
||||
}
|
||||
|
||||
###
|
||||
# 3. Responses API — continue the conversation by previous_response_id
|
||||
# Replace <RESPONSE_ID> with `id` from one of the responses above —
|
||||
# the host uses it as the workflow checkpoint scope key, so the
|
||||
# workflow resumes from its latest checkpoint before applying the
|
||||
# new input.
|
||||
POST {{host}}/responses
|
||||
Content-Type: application/json
|
||||
|
||||
{
|
||||
"model": "agent",
|
||||
"previous_response_id": "<RESPONSE_ID>",
|
||||
"input": "{\"topic\": \"electric SUV\", \"style\": \"retro\", \"audience\": \"boomers\"}"
|
||||
}
|
||||
|
||||
###
|
||||
# 4. Readiness probe
|
||||
GET {{host}}/readiness
|
||||
@@ -1,23 +0,0 @@
|
||||
[project]
|
||||
name = "agent-framework-hosting-sample-local-responses-workflow"
|
||||
version = "0.0.1"
|
||||
description = "Local hosting sample exposing a 3-agent workflow over the Responses API with per-conversation checkpoint storage."
|
||||
requires-python = ">=3.10"
|
||||
dependencies = [
|
||||
"agent-framework-foundry",
|
||||
"agent-framework-hosting",
|
||||
"agent-framework-hosting-responses",
|
||||
"azure-identity",
|
||||
"aiohttp>=3.13.5",
|
||||
"hypercorn>=0.17",
|
||||
]
|
||||
|
||||
[dependency-groups]
|
||||
dev = ["openai>=1.99"]
|
||||
|
||||
[tool.uv]
|
||||
package = false
|
||||
|
||||
[tool.uv.sources]
|
||||
agent-framework-hosting = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting" }
|
||||
agent-framework-hosting-responses = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting-responses" }
|
||||
Generated
-62
@@ -46,8 +46,6 @@ members = [
|
||||
"agent-framework-foundry-local",
|
||||
"agent-framework-gemini",
|
||||
"agent-framework-github-copilot",
|
||||
"agent-framework-hosting",
|
||||
"agent-framework-hosting-responses",
|
||||
"agent-framework-hyperlight",
|
||||
"agent-framework-lab",
|
||||
"agent-framework-mem0",
|
||||
@@ -626,57 +624,6 @@ requires-dist = [
|
||||
{ name = "github-copilot-sdk", marker = "python_full_version >= '3.11'", specifier = "==1.0.2" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "agent-framework-hosting"
|
||||
version = "1.0.0a260424"
|
||||
source = { editable = "packages/hosting" }
|
||||
dependencies = [
|
||||
{ name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
{ name = "starlette", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
]
|
||||
|
||||
[package.optional-dependencies]
|
||||
disk = [
|
||||
{ name = "diskcache", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
]
|
||||
serve = [
|
||||
{ name = "hypercorn", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
]
|
||||
|
||||
[package.dev-dependencies]
|
||||
dev = [
|
||||
{ name = "httpx", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
]
|
||||
|
||||
[package.metadata]
|
||||
requires-dist = [
|
||||
{ name = "agent-framework-core", editable = "packages/core" },
|
||||
{ name = "diskcache", marker = "extra == 'disk'", specifier = ">=5.6" },
|
||||
{ name = "hypercorn", marker = "extra == 'serve'", specifier = ">=0.17" },
|
||||
{ name = "starlette", specifier = ">=0.37" },
|
||||
]
|
||||
provides-extras = ["serve", "disk"]
|
||||
|
||||
[package.metadata.requires-dev]
|
||||
dev = [{ name = "httpx", specifier = ">=0.28.1" }]
|
||||
|
||||
[[package]]
|
||||
name = "agent-framework-hosting-responses"
|
||||
version = "1.0.0a260424"
|
||||
source = { editable = "packages/hosting-responses" }
|
||||
dependencies = [
|
||||
{ name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
{ name = "agent-framework-hosting", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
{ name = "openai", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
]
|
||||
|
||||
[package.metadata]
|
||||
requires-dist = [
|
||||
{ name = "agent-framework-core", editable = "packages/core" },
|
||||
{ name = "agent-framework-hosting", editable = "packages/hosting" },
|
||||
{ name = "openai", specifier = ">=1.99.0,<3" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "agent-framework-hyperlight"
|
||||
version = "1.0.0b260521"
|
||||
@@ -2306,15 +2253,6 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/c7/26/4a2bad8eb430d8d805a4642c4bff25103a37548d74ab346f8b1e024abcc5/deepdiff-9.1.0-py3-none-any.whl", hash = "sha256:80c0460e1993b04f6f0ca79abf25548b129fd218478c4ebb08f80560f5d10610", size = 184662 },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "diskcache"
|
||||
version = "5.6.3"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/3f/21/1c1ffc1a039ddcc459db43cc108658f32c57d271d7289a2794e401d0fdb6/diskcache-5.6.3.tar.gz", hash = "sha256:2c3a3fa2743d8535d832ec61c2054a1641f41775aa7c556758a109941e33e4fc", size = 67916, upload-time = "2023-08-31T06:12:00.316Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/3f/27/4570e78fc0bf5ea0ca45eb1de3818a23787af9b390c0b0a0033a1b8236f9/diskcache-5.6.3-py3-none-any.whl", hash = "sha256:5e31b2d5fbad117cc363ebaf6b689474db18a1f6438bc82358b024abd4c2ca19", size = 45550, upload-time = "2023-08-31T06:11:58.822Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "distro"
|
||||
version = "1.9.0"
|
||||
|
||||
Reference in New Issue
Block a user