Decorators Reference

Author tools and workflows with MCPApp decorators.

Decorators

MCPApp exposes a small set of decorators that register tools, workflows, and workflow tasks. The decorators are engine-aware: when you switch from the default asyncio executor to Temporal, the same annotations automatically apply the appropriate Temporal SDK wrappers.

Decorator	Applies to	Purpose
@app.tool	sync function	Exposes a blocking function as an MCP tool.
@app.async_tool	async function	Registers an async tool and publishes it through FastMCP.
@app.workflow	Workflow subclass	Declares a workflow class and adapts it for the active execution engine.
@app.workflow_run	async method	Marks the primary run coroutine for a workflow.
@app.workflow_task	async function/method	Registers an activity task reusable across workflows.
@app.workflow_signal	async method	Handles inbound workflow signals (Temporal-friendly).

Tool decorators

Tools expose code as MCP functions that agents and LLMs can call. Both decorators accept the same keyword arguments:

Parameter	Description
name	Override the exported MCP tool name (defaults to the function name).
title	Short display label for clients.
description	Human-readable description; the function docstring is used when omitted.
annotations	Supply a ToolAnnotations object or mapping for MCP metadata.
icons	Provide one or more Icon instances or mappings for client rendering.
meta	Arbitrary metadata forwarded to FastMCP.
structured_output	Set to True to hint that the result is structured JSON; some LLMs will choose schema-aware models automatically.

@app.tool — synchronous tools

from pydantic import BaseModel
from mcp_agent.app import MCPApp

app = MCPApp(name="reporting")

class Summary(BaseModel):
    title: str
    verdict: str

@app.tool(description="Summarise a raw document into a headline and verdict.", structured_output=True)
def summarise_document(text: str) -> Summary:
    """Summarise content deterministically."""
    title = text.splitlines()[0][:120]
    verdict = "APPROVED" if "ship it" in text.lower() else "NEEDS REVIEW"
    return Summary(title=title, verdict=verdict)

• The decorator validates the signature up-front; missing type hints or unsupported default values raise an error during import.
• @app.tool automatically creates a hidden workflow so the tool is reachable via both callTool and the workflow endpoints (run, get_status) exposed by FastMCP.
• The function executes inside the app event loop; heavy work should offload itself (for example, using asyncio.to_thread).

@app.async_tool — asynchronous tools

import httpx

@app.async_tool(
    name="fetch_page",
    description="Fetch raw HTML from an HTTP endpoint.",
    icons=[{"emoji": "🌐"}],
)
async def fetch_page(url: str, *, timeout: int = 20) -> str:
    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.text

• The coroutine is awaited directly, so you can call other async APIs without wrappers.
• When Temporal is active, the decorated function is wrapped with workflow.activity metadata automatically.

Workflow decorator suite

Workflows orchestrate complex sequences, combining tasks, tools, and signals. Every workflow subclass must inherit from mcp_agent.executor.workflow.Workflow.

from datetime import timedelta
from mcp_agent.executor.workflow import Workflow, WorkflowResult

@app.workflow
class PublishArticle(Workflow[WorkflowResult[str]]):
    @app.workflow_task(schedule_to_close_timeout=timedelta(minutes=5))
    async def generate_outline(self, topic: str) -> str:
        outline = f"- Introduction to {topic}\n- Key findings\n- Next steps"
        self.context.logger.info("Generated outline", data={"topic": topic})
        return outline

    @app.workflow_task(retry_policy={"maximum_attempts": 3})
    async def send_to_editor(self, outline: str) -> None:
        self.context.logger.info("Sending to editor", data={"outline": outline})

    @app.workflow_signal(name="editor_feedback")
    async def editor_feedback(self, notes: str) -> None:
        self.state.feedback = notes or ""

    @app.workflow_run
    async def run(self, topic: str) -> WorkflowResult[str]:
        outline = await self.generate_outline(topic)
        await self.send_to_editor(outline)
        feedback = getattr(self.state, "feedback", "Awaiting review.")
        return WorkflowResult(value=f"{outline}\n\nEditor feedback: {feedback}")

@app.workflow

• Registers the class with the app and applies engine-specific decorators (workflow.defn for Temporal, no-op for asyncio).
• An optional workflow_id parameter lets you export the workflow under a different name when registering.
• The decorator stores a reference to the MCPApp, letting workflow instances access self.context.app.

@app.workflow_run

Wraps the run coroutine so that initialization, tracing, and Temporal-specific instrumentation are handled automatically. You rarely need to call it manually—applying @app.workflow and naming the method run is enough—but explicit usage lets you decorate additional entry points.

@app.workflow_task

Registers a coroutine as a reusable activity. Key options:

Option	Effect
name	Override the fully qualified activity name (defaults to <module>.<qualname>).
schedule_to_close_timeout	Maximum wall-clock duration allowed for the task.
retry_policy	Temporal retry configuration (e.g. {"maximum_attempts": 5}).
**meta_kwargs	Arbitrary metadata stored alongside the task for inspection.

The decorator enforces that the target is async; synchronous functions should wrap their blocking work with asyncio.to_thread.

Tasks defined outside a workflow are also supported—they are registered globally and can be reused across multiple workflows.

@app.workflow_signal

Signals let external actors (humans, webhooks, other workflows) nudge a running workflow. The decorator accepts an optional name argument and works in both asyncio and Temporal modes.

@app.workflow_signal(name="user_input")
async def collect_user_input(self, payload: str) -> None:
    self.context.logger.info("Received payload", data={"payload": payload})
    self.state.latest_payload = payload

The generated wrapper automatically strips the workflow instance (self) for Temporal’s signal handler signature.

All workflow decorators defer to the active executor. When you switch to Temporal, tasks become activities, run becomes a workflow entry point, and signals map to @workflow.signal—no additional changes required.

Putting it together

from mcp_agent.app import MCPApp
from mcp_agent.executor.workflow import Workflow

app = MCPApp(name="insight-generator")

@app.tool(description="List the MCP servers registered with this app.")
def list_servers() -> list[str]:
    return sorted((app.config.mcp.servers or {}).keys())

@app.workflow
class ResearchWorkflow(Workflow[None]):
    @app.workflow_task()
    async def gather_sources(self, query: str) -> list[str]:
        self.context.logger.info("Gathering sources", data={"query": query})
        return [f"https://example.com/search?q={query}"]

    @app.workflow_run
    async def run(self, topic: str) -> None:
        sources = await self.gather_sources(topic)
        self.context.logger.info("Workflow completed", data={"topic": topic, "sources": sources})

This pattern gives you:

• A reusable list_servers tool that exposes runtime metadata without boilerplate.
• A workflow that can run locally (asyncio) or durably (Temporal) with the same code.
• Optional signals/tasks to pause, resume, or branch as needed.

For CLI-driven workflows and deployment details, continue with the CLI reference. For configuration options that pair with these decorators, review the Configuration reference.