chatomon.connectors.maf ¶

Build an atom into a native Microsoft Agent Framework agent.

Install with pip install 'chatomon[maf]' or chatomon[maf-foundry].

Classes:

Name	Description
`MAFConnector`	Build an atom into a native `agent_framework.Agent`.
`MAFFoundryConnector`	Build an atom into a native `agent_framework.foundry.FoundryAgent`.

MAFConnector ¶

MAFConnector(
    client: Any,
    *,
    resolve_client: Callable[[str], Any] | None = None,
    **agent_kwargs: Any
)

Bases: Connector[Agent[Any]]

Build an atom into a native agent_framework.Agent.

The client can be any MAF chat client that implements ChatMiddlewareLayer, including FoundryChatClient. Provider clients such as OpenAIChatClient need their own MAF distribution in addition to chatomon[maf]. The connector places chatomon middleware outside agent middleware supplied by the caller. Local AgentSession history is used with store=False; pass a session from agent.create_session() to keep turns together. MAF's native run API is asynchronous.

Parameters:

Name	Type	Description	Default
`client`	`Any`	The nucleus MAF chat client.	required
`resolve_client`	`Callable[[str], Any] \| None`	Maps a `JudgeOrbital` model ID to another MAF chat client.	`None`
`**agent_kwargs`	`Any`	Forwarded to `agent_framework.Agent`.	`{}`

Raises:

Type	Description
`ChatomonConfigError`	If an argument conflicts with atom identity, enables remote conversation continuity, or bypasses the chat middleware.

Examples:

import asyncio
import os

from agent_framework.openai import OpenAIChatClient
from openai import AsyncOpenAI

from chatomon import Atom
from chatomon.connectors.maf import MAFConnector


async def main() -> None:
    async with AsyncOpenAI() as provider:
        client = OpenAIChatClient(
            model=os.environ["MAF_MODEL"], async_client=provider
        )
        agent = Atom(
            name="assistant",
            instructions="Be concise.",
            connector=MAFConnector(client),
        ).build()
        reply = await agent.run("Hello", session=agent.create_session())
        print(reply.text)


asyncio.run(main())

Methods:

Name	Description
`build`	Build the native MAF agent with chatomon's middleware installed.
`generate_structured`	Run a structured judge call with a resolved MAF chat client.

Source code in src/chatomon/connectors/maf/_core.py

def __init__(
    self,
    client: Any,
    *,
    resolve_client: Callable[[str], Any] | None = None,
    **agent_kwargs: Any,
) -> None:
    self.client = client
    self.resolve_client = resolve_client
    self.agent_kwargs = _prepare_kwargs(agent_kwargs)

build ¶

build(atom: Atom[Agent[Any]], engine: Engine) -> Agent[Any]

Build the native MAF agent with chatomon's middleware installed.

Parameters:

Name	Type	Description	Default
`atom`	`Atom[Agent[Any]]`	The validated atom.	required
`engine`	`Engine`	The engine that runs the atom's orbitals.	required

Returns:

Type	Description
`Agent[Any]`	A native MAF agent.

Raises:

Type	Description
`ChatomonConfigError`	If the client cannot run chat middleware or preconfigured client middleware could rewrite checked output.

Source code in src/chatomon/connectors/maf/_core.py

def build(self, atom: Atom[Agent[Any]], engine: Engine) -> Agent[Any]:
    """Build the native MAF agent with chatomon's middleware installed.

    Args:
        atom: The validated atom.
        engine: The engine that runs the atom's orbitals.

    Returns:
        A native MAF agent.

    Raises:
        ChatomonConfigError: If the client cannot run chat middleware or
            preconfigured client middleware could rewrite checked output.
    """
    _require_resolver(atom, self, self.resolve_client)
    if not isinstance(self.client, ChatMiddlewareLayer):
        raise ChatomonConfigError(
            "MAFConnector requires a client with ChatMiddlewareLayer so chatomon "
            "can check replies before local history is saved"
        )
    if engine.output_orbitals and self.client.chat_middleware:
        raise ChatomonConfigError(
            "preconfigured client chat middleware can rewrite checked output; "
            "move it to MAFConnector(middleware=...)"
        )
    kwargs = dict(self.agent_kwargs)
    middleware = [
        _ChatomonAgentMiddleware(engine, atom.blocked_placeholder),
        _ChatomonChatMiddleware(engine),
        *(kwargs.pop("middleware", ()) or ()),
    ]
    return Agent(
        client=cast(Any, self.client),
        name=atom.name,
        instructions=atom.system_prompt,
        middleware=middleware,
        **kwargs,
    )

generate_structured `async` ¶

generate_structured(
    model: str,
    *,
    system: str,
    prompt: str,
    schema: type[ModelT]
) -> ModelT

Run a structured judge call with a resolved MAF chat client.

Parameters:

Name	Type	Description	Default
`model`	`str`	The judge model identifier.	required
`system`	`str`	Judge instructions.	required
`prompt`	`str`	Judge input.	required
`schema`	`type[ModelT]`	Required Pydantic result type.	required

Returns:

Type	Description
`ModelT`	The validated judge result.

Raises:

Type	Description
`ChatomonConfigError`	If `resolve_client` is missing or invalid.
`ValueError`	If the judge returned no structured answer.

Source code in src/chatomon/connectors/maf/_core.py

async def generate_structured(
    self, model: str, *, system: str, prompt: str, schema: type[ModelT]
) -> ModelT:
    """Run a structured judge call with a resolved MAF chat client.

    Args:
        model: The judge model identifier.
        system: Judge instructions.
        prompt: Judge input.
        schema: Required Pydantic result type.

    Returns:
        The validated judge result.

    Raises:
        ChatomonConfigError: If `resolve_client` is missing or invalid.
        ValueError: If the judge returned no structured answer.
    """  # noqa: DOC502 - delegated to _generate_structured
    return await _generate_structured(
        self.resolve_client, model, system=system, prompt=prompt, schema=schema
    )

MAFFoundryConnector ¶

MAFFoundryConnector(
    *,
    resolve_client: Callable[[str], Any] | None = None,
    **foundry_kwargs: Any
)

Bases: Connector['FoundryAgent']

Build an atom into a native agent_framework.foundry.FoundryAgent.

This connector attaches to an agent already configured in Foundry. Input and output orbitals run around its local MAF session. Atom instructions and knowledge are sent as a developer message because Foundry rejects the request-level instructions field when an agent is specified. Rejected drafts are replaced or regenerated before entering local history.

Parameters:

Name	Type	Description	Default
`resolve_client`	`Callable[[str], Any] \| None`	Maps a `JudgeOrbital` model ID to a MAF chat client.	`None`
`**foundry_kwargs`	`Any`	Forwarded to `FoundryAgent`; include the project endpoint, agent name and credential or project client. `client_type` is managed by this connector.	`{}`

Raises:

Type	Description
`ChatomonConfigError`	If an argument conflicts with atom identity or enables remote conversation continuity.
`MissingExtraError`	If `chatomon[maf-foundry]` is not installed.

Examples:

import asyncio
import os

from azure.identity.aio import DefaultAzureCredential

from chatomon import Atom
from chatomon.connectors.maf import MAFFoundryConnector


async def main() -> None:
    async with DefaultAzureCredential() as credential:
        agent = Atom(
            name="assistant",
            instructions="Be concise.",
            connector=MAFFoundryConnector(
                project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
                agent_name=os.environ["FOUNDRY_AGENT_NAME"],
                agent_version=os.environ["FOUNDRY_AGENT_VERSION"],
                credential=credential,
            ),
        ).build()
        try:
            reply = await agent.run("Hello", session=agent.create_session())
            print(reply.text)
        finally:
            await agent.client.close()


asyncio.run(main())

Methods:

Name	Description
`build`	Build a native Foundry agent with checks and local history.
`generate_structured`	Run a structured judge call with a resolved MAF chat client.

Source code in src/chatomon/connectors/maf/_foundry.py

def __init__(
    self,
    *,
    resolve_client: Callable[[str], Any] | None = None,
    **foundry_kwargs: Any,
) -> None:
    if "client_type" in foundry_kwargs:
        raise ChatomonConfigError(
            "MAFFoundryConnector sets client_type to preserve atom instructions "
            "and chat middleware"
        )
    self.resolve_client = resolve_client
    self.foundry_kwargs = _prepare_kwargs(foundry_kwargs)

build ¶

build(
    atom: Atom[FoundryAgent], engine: Engine
) -> FoundryAgent

Build a native Foundry agent with checks and local history.

Parameters:

Name	Type	Description	Default
`atom`	`Atom[FoundryAgent]`	The validated atom.	required
`engine`	`Engine`	The engine that runs the atom's orbitals.	required

Returns:

Type	Description
`FoundryAgent`	A native Foundry agent.

Raises:

Type	Description
`ChatomonConfigError`	If a custom client type is supplied.
`MissingExtraError`	If the Foundry extra is not installed.

Source code in src/chatomon/connectors/maf/_foundry.py

def build(self, atom: Atom[FoundryAgent], engine: Engine) -> FoundryAgent:
    """Build a native Foundry agent with checks and local history.

    Args:
        atom: The validated atom.
        engine: The engine that runs the atom's orbitals.

    Returns:
        A native Foundry agent.

    Raises:
        ChatomonConfigError: If a custom client type is supplied.
        MissingExtraError: If the Foundry extra is not installed.
    """
    _require_resolver(atom, self, self.resolve_client)
    try:
        from agent_framework.foundry import FoundryAgent as NativeFoundryAgent

        from chatomon.connectors.maf._foundry_client import _ChatomonFoundryChatClient
    except ImportError as exc:
        raise MissingExtraError("maf-foundry", "agent-framework-foundry") from exc
    kwargs = dict(self.foundry_kwargs)
    middleware = [
        _ChatomonAgentMiddleware(engine, atom.blocked_placeholder),
        _ChatomonChatMiddleware(engine, foundry_instructions=atom.system_prompt),
        *(kwargs.pop("middleware", ()) or ()),
    ]
    agent = NativeFoundryAgent(
        name=atom.name,
        instructions=atom.system_prompt,
        client_type=_ChatomonFoundryChatClient,
        middleware=middleware,
        **kwargs,
    )
    if not isinstance(agent.client, ChatMiddlewareLayer):
        raise ChatomonConfigError(
            "MAFFoundryConnector requires a Foundry client with ChatMiddlewareLayer"
        )
    return agent

generate_structured `async` ¶

generate_structured(
    model: str,
    *,
    system: str,
    prompt: str,
    schema: type[ModelT]
) -> ModelT

Run a structured judge call with a resolved MAF chat client.

Parameters:

Name	Type	Description	Default
`model`	`str`	The judge model identifier.	required
`system`	`str`	Judge instructions.	required
`prompt`	`str`	Judge input.	required
`schema`	`type[ModelT]`	Required Pydantic result type.	required

Returns:

Type	Description
`ModelT`	The validated judge result.

Raises:

Type	Description
`ChatomonConfigError`	If `resolve_client` is missing or invalid.
`ValueError`	If the judge returned no structured answer.

Source code in src/chatomon/connectors/maf/_foundry.py

async def generate_structured(
    self, model: str, *, system: str, prompt: str, schema: type[ModelT]
) -> ModelT:
    """Run a structured judge call with a resolved MAF chat client.

    Args:
        model: The judge model identifier.
        system: Judge instructions.
        prompt: Judge input.
        schema: Required Pydantic result type.

    Returns:
        The validated judge result.

    Raises:
        ChatomonConfigError: If `resolve_client` is missing or invalid.
        ValueError: If the judge returned no structured answer.
    """  # noqa: DOC502 - delegated to _generate_structured
    return await _generate_structured(
        self.resolve_client, model, system=system, prompt=prompt, schema=schema
    )

chatomon.connectors.maf ¶

MAFConnector ¶

build ¶

generate_structured async ¶

MAFFoundryConnector ¶

build ¶

generate_structured async ¶

generate_structured `async` ¶

generate_structured `async` ¶