Agent 与工具（2/8）

        # Read from Store: get user's enabled features
        store = request.runtime.store
        feature_flags = store.get(("features",), user_id)

        if feature_flags:
            enabled_features = feature_flags.value.get("enabled_tools", [])
            # Only include tools that are enabled for this user
            tools = [t for t in request.tools if t.name in enabled_features]
            request = request.override(tools=tools)

        return handler(request)

    agent = create_agent(
        model="gpt-5.5",
        tools=[search_tool, analysis_tool, export_tool],
        middleware=[store_based_tools],
        context_schema=Context,
        store=InMemoryStore()
    )




Filter tools based on user permissions from Runtime Context:

```python
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable

@dataclass
class Context:
    user_role: str

@wrap_model_call
def context_based_tools(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """Filter tools based on Runtime Context permissions."""
    # Read from Runtime Context: get user role
    if request.runtime is None or request.runtime.context is None:
        # If no context provided, default to viewer (most restrictive)
        user_role = "viewer"
    else:
        user_role = request.runtime.context.user_role

    if user_role == "admin":
        # Admins get all tools
        pass
    elif user_role == "editor":
        # Editors can't delete
        tools = [t for t in request.tools if t.name != "delete_data"]
        request = request.override(tools=tools)
    else:
        # Viewers get read-only tools
        tools = [t for t in request.tools if t.name.startswith("read_")]
        request = request.override(tools=tools)

    return handler(request)

agent = create_agent(
    model="gpt-5.5",
    tools=[read_data, write_data, delete_data],
    middleware=[context_based_tools],
    context_schema=Context
)

This approach is best when:

* All possible tools are known at compile/startup time
* You want to filter based on permissions, feature flags, or conversation state
* Tools are static but their availability is dynamic

See [Dynamically selecting tools](https://docs.langchain.com/oss/python/langchain/middleware/custom#dynamically-selecting-tools) for more examples.



When tools are discovered or created at runtime (e.g., loaded from an MCP server, generated based on user data, or fetched from a remote registry), you need to both register the tools and handle their execution dynamically.

This requires two middleware hooks:

1. `wrap_model_call` - Add the dynamic tools to the request
2. `wrap_tool_call` - Handle execution of the dynamically added tools

from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelRequest, ToolCallRequest

# A tool that will be added dynamically at runtime
@tool
def calculate_tip(bill_amount: float, tip_percentage: float = 20.0) -> str:
    """Calculate the tip amount for a bill."""
    tip = bill_amount * (tip_percentage / 100)
    return f"Tip: ${tip:.2f}, Total: ${bill_amount + tip:.2f}"

class DynamicToolMiddleware(AgentMiddleware):
    """Middleware that registers and handles dynamic tools."""

    def wrap_model_call(self, request: ModelRequest, handler):
        # Add dynamic tool to the request
        # This could be loaded from an MCP server, database, etc.
        updated = request.override(tools=[*request.tools, calculate_tip])
        return handler(updated)

    def wrap_tool_call(self, request: ToolCallRequest, handler):
        # Handle execution of the dynamic tool
        if request.tool_call["name"] == "calculate_tip":
            return handler(request.override(tool=calculate_tip))
        return handler(request)

agent = create_agent(
    model="gpt-5.5",
    tools=[get_weather],  # Only static tools registered here
    middleware=[DynamicToolMiddleware()],
)

# The agent can now use both get_weather AND calculate_tip
result = agent.invoke({
    "messages": [{"role": "user", "content": "Calculate a 20% tip on $85"}]
})

This approach is best when:

* Tools are discovered at runtime (e.g., from an MCP server)
* Tools are generated dynamically based on user data or configuration
* You're integrating with external tool registries


  The `wrap_tool_call` hook is required for runtime-registered tools because the agent needs to know how to execute tools that weren't in the original tool list. Without it, the agent won't know how to invoke the dynamically added tool.

Headless tools

Some tools should run where your user's app runs (typically the browser), not inside the process. Headless tools are tool definitions, which include the name, description, and argument schema, that you register on the server with your agent. The implementation is registered only on the client and executed after a short interrupt/resume handshake.

This is different from ordinary tools whose function body runs on the server, and from server-side tool use where the model provider executes built-in tools remotely.

When to use headless tools

Use them when the work depends on the environment, device, or UI that only exists on the client. For example:

Browser APIs: Geolocation, IndexedDB, Clipboard, Canvas 2D, file pickers, Battery API, etc.
Privacy and locality: Data stays on the device (for example, local “memory” in IndexedDB).
Latency: No extra server round trip for purely local operations.
Structured, safe effects: Prefer many small, typed tools (for example one tool per canvas primitive) instead of sending arbitrary code to eval.

How the pattern works

In both runtimes, the model sees a normal tool it can call, but the actual execution happens outside the server process.

Define a headless tool with tool(name=..., description=..., args_schema=...) from langchain.tools. A headless tool is schema-only, with no in-process implementation.
Register that tool with create_agent or your LangGraph graph so the model can call it normally.
Handle the interrupt payload when the tool is invoked. Instead of running locally, the graph pauses with a payload shaped like {"type": "tool", "tool_call": {"id", "name", "args"}}.
Resume the graph after your app, another service, or a human step performs the action. For browser-based flows, you can mirror the schema in the frontend and attach .implement(...) there.

If you call tool(...) in Python with only name, description, and args_schema, LangChain returns a HeadlessTool. There is no .implement() API on the Python side.

When the model issues a tool call for one of these tools, the run interrupts instead of executing the tool locally. Your app can inspect the payload, perform the action in the right environment (for example a browser, another service, or a human review step), then resume the graph with the tool result. When you use the supported JS SDK hooks, they can detect headless-tool interrupts, run the matching client implementation, and submit the resume command for you.

Use the optional onTool callback to observe lifecycle events (start, success, error) for UI feedback such as spinners or toasts.

See an end-to-end example of schema-only tools executed in the client with useStream.

Prebuilt tools

LangChain provides a large collection of prebuilt tools and toolkits for common tasks like web search, code interpretation, database access, and more. These ready-to-use tools can be directly integrated into your agents without writing custom code.

See the tools and toolkits integration page for a complete list of available tools organized by category.

Tools from MCP servers

Model Context Protocol (MCP) is an open protocol that standardizes how applications expose tools to language models. Rather than hand-authoring a tool, you connect to an MCP server and adapt the tools it advertises into LangChain tools, ready to pass to an agent like any other tool.

MCPAdapter discovers a server's tools and converts them into LangChain tools. Open the adapter, call list_tools(), and hand the result to create_agent:

from langchain.agents import create_agent
from langchain.mcp import MCPAdapter

async with MCPAdapter("https://example.com/mcp") as adapter:
    tools = await adapter.list_tools()
    agent = create_agent("claude-sonnet-4-6", tools)

The langchain.mcp namespace requires langchain[mcp]>=1.4.0 and is in beta. The API may change.

For transports, authentication, multiple servers, and handling tool results, see Model Context Protocol (MCP).

Server-side tool use

Some chat models feature built-in tools that are executed server-side by the model provider. These include capabilities like web search and code interpreters that don't require you to define or host the tool logic.

Refer to the individual chat model integration pages and the tool calling documentation for details on enabling and using these built-in tools.

[Connect these docs](https://docs.langchain.com/use-these-docs) to your agent of choice via MCP for real-time answers. [Edit this page on GitHub](https://github.com/langchain-ai/docs/edit/main/src/oss/langchain/tools.mdx) or [file an issue](https://github.com/langchain-ai/docs/issues/new/choose).

Tables

customers - customer_id (PRIMARY KEY) - name - email - signup_date - status (active/inactive) - customer_tier (bronze/silver/gold/platinum)

orders - order_id (PRIMARY KEY) - customer_id (FOREIGN KEY -> customers) - order_date - status (pending/completed/cancelled/refunded) - total_amount - sales_region (north/south/east/west)

[... rest of schema ...]

Business Logic

High-value orders: Orders with total_amount > 1000 Revenue calculation: Only count orders with status = 'completed'

================================== Ai Message ==================================

Here's a SQL query to find all customers who made orders over $1000 in the last month:

```sql SELECT DISTINCT c.customer_id, c.name, c.email, c.customer_tier FROM customers c JOIN orders o ON c.customer_id = o.customer_id WHERE o.total_amount > 1000 AND o.status = 'completed' AND o.order_date >= CURRENT_DATE - INTERVAL '1 month' ORDER BY c.customer_id; ```

This query: - Joins customers with their orders - Filters for high-value orders (>$1000) using the total_amount field - Only includes completed orders (as per the business logic) - Restricts to orders from the last month - Returns distinct customers to avoid duplicates if they made multiple qualifying orders


The agent saw the lightweight skill description in its system prompt, recognized the question required sales database knowledge, called `load_skill("sales_analytics")` to get the full schema and business logic, and then used that information to write a correct query following the database conventions.

## 6. Advanced: Add constraints with custom state

  You can add constraints to enforce that certain tools are only available after specific skills have been loaded. This requires tracking which skills have been loaded in custom agent state.

  ### Define custom state

  First, extend the agent state to track loaded skills:

  ```python
  from langchain.agents.middleware import AgentState

  class CustomState(AgentState):  # [!code highlight]
      skills_loaded: NotRequired[list[str]]  # Track which skills have been loaded  # [!code highlight]

### Update load_skill to modify state

Modify the load_skill tool to update state when a skill is loaded:

from langgraph.types import Command  # [!code highlight]
from langchain.tools import tool, ToolRuntime
from langchain.messages import ToolMessage  # [!code highlight]

@tool
def load_skill(skill_name: str, runtime: ToolRuntime) -> Command:  # [!code highlight]
    """Load the full content of a skill into the agent's context.

    Use this when you need detailed information about how to handle a specific
    type of request. This will provide you with comprehensive instructions,
    policies, and guidelines for the skill area.

    Args:
        skill_name: The name of the skill to load
    """
    # Find and return the requested skill
    for skill in SKILLS:
        if skill["name"] == skill_name:
            skill_content = f"Loaded skill: {skill_name}\n\n{skill['content']}"

            # Update state to track loaded skill
            return Command(  # [!code highlight]
                update={  # [!code highlight]
                    "messages": [  # [!code highlight]
                        ToolMessage(  # [!code highlight]
                            content=skill_content,  # [!code highlight]
                            tool_call_id=runtime.tool_call_id,  # [!code highlight]
                        )  # [!code highlight]
                    ],  # [!code highlight]
                    "skills_loaded": [skill_name],  # [!code highlight]
                }  # [!code highlight]
            )  # [!code highlight]

    # Skill not found
    available = ", ".join(s["name"] for s in SKILLS)
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"Skill '{skill_name}' not found. Available skills: {available}",
                    tool_call_id=runtime.tool_call_id,
                )
            ]
        }
    )

### Create constrained tool

Create a tool that's only usable after a specific skill has been loaded:

@tool
def write_sql_query(  # [!code highlight]
    query: str,
    vertical: str,
    runtime: ToolRuntime,
) -> str:
    """Write and validate a SQL query for a specific business vertical.

    This tool helps format and validate SQL queries. You must load the
    appropriate skill first to understand the database schema.

    Args:
        query: The SQL query to write
        vertical: The business vertical (sales_analytics or inventory_management)
    """
    # Check if the required skill has been loaded
    skills_loaded = runtime.state.get("skills_loaded", [])  # [!code highlight]

    if vertical not in skills_loaded:  # [!code highlight]
        return (  # [!code highlight]
            f"Error: You must load the '{vertical}' skill first "  # [!code highlight]
            f"to understand the database schema before writing queries. "  # [!code highlight]
            f"Use load_skill('{vertical}') to load the schema."  # [!code highlight]
        )  # [!code highlight]

    # Validate and format the query
    return (
        f"SQL Query for {vertical}:\n\n"
        f"```sql\n{query}\n```\n\n"
        f"✓ Query validated against {vertical} schema\n"
        f"Ready to execute against the database."
    )

Update middleware and agent

Update the middleware to use the custom state schema:

class SkillMiddleware(AgentMiddleware[CustomState]):  # [!code highlight]
    """Middleware that injects skill descriptions into the system prompt."""

    state_schema = CustomState  # [!code highlight]
    tools = [load_skill, write_sql_query]  # [!code highlight]

    # ... rest of the middleware implementation stays the same

Create the agent with the middleware that registers the constrained tool:

agent = create_agent(
    model,
    system_prompt=(
        "You are a SQL query assistant that helps users "
        "write queries against business databases."
    ),
    middleware=[SkillMiddleware()],  # [!code highlight]
    checkpointer=InMemorySaver(),
)

Now if the agent tries to use write_sql_query before loading the required skill, it will receive an error message prompting it to load the appropriate skill (e.g., sales_analytics or inventory_management) first. This ensures the agent has the necessary schema knowledge before attempting to validate queries.

Complete example

Here's a complete, runnable implementation combining all the pieces from this tutorial:

from langchain_core.utils.uuid import uuid7
from typing import TypedDict, NotRequired
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import ModelRequest, ModelResponse, AgentMiddleware
from langchain.messages import SystemMessage
from langgraph.checkpoint.memory import InMemorySaver
from typing import Callable

# Define skill structure
class Skill(TypedDict):
    """A skill that can be progressively disclosed to the agent."""
    name: str
    description: str
    content: str

# Define skills with schemas and business logic
SKILLS: list[Skill] = [
    {
        "name": "sales_analytics",
        "description": "Database schema and business logic for sales data analysis including customers, orders, and revenue.",
        "content": """# Sales Analytics Schema

## Tables

### customers
- customer_id (PRIMARY KEY)
- name
- email
- signup_date
- status (active/inactive)
- customer_tier (bronze/silver/gold/platinum)

### orders
- order_id (PRIMARY KEY)
- customer_id (FOREIGN KEY -> customers)
- order_date
- status (pending/completed/cancelled/refunded)
- total_amount
- sales_region (north/south/east/west)

### order_items
- item_id (PRIMARY KEY)
- order_id (FOREIGN KEY -> orders)
- product_id
- quantity
- unit_price
- discount_percent

## Business Logic

**Active customers**: status = 'active' AND signup_date <= CURRENT_DATE - INTERVAL '90 days'

**Revenue calculation**: Only count orders with status = 'completed'. Use total_amount from orders table, which already accounts for discounts.

**Customer lifetime value (CLV)**: Sum of all completed order amounts for a customer.

**High-value orders**: Orders with total_amount > 1000

## Example Query

-- Get top 10 customers by revenue in the last quarter
SELECT
    c.customer_id,
    c.name,
    c.customer_tier,
    SUM(o.total_amount) as total_revenue
FROM customers c
JOIN orders o ON c.customer_id = o.customer_id
WHERE o.status = 'completed'
  AND o.order_date >= CURRENT_DATE - INTERVAL '3 months'
GROUP BY c.customer_id, c.name, c.customer_tier
ORDER BY total_revenue DESC
LIMIT 10;
""",
    },
    {
        "name": "inventory_management",
        "description": "Database schema and business logic for inventory tracking including products, warehouses, and stock levels.",
        "content": """# Inventory Management Schema

## Tables

### products
- product_id (PRIMARY KEY)
- product_name
- sku
- category
- unit_cost
- reorder_point (minimum stock level before reordering)
- discontinued (boolean)

### warehouses
- warehouse_id (PRIMARY KEY)
- warehouse_name
- location
- capacity

### inventory
- inventory_id (PRIMARY KEY)
- product_id (FOREIGN KEY -> products)
- warehouse_id (FOREIGN KEY -> warehouses)
- quantity_on_hand
- last_updated

### stock_movements
- movement_id (PRIMARY KEY)
- product_id (FOREIGN KEY -> products)
- warehouse_id (FOREIGN KEY -> warehouses)
- movement_type (inbound/outbound/transfer/adjustment)
- quantity (positive for inbound, negative for outbound)
- movement_date
- reference_number

## Business Logic

**Available stock**: quantity_on_hand from inventory table where quantity_on_hand > 0

**Products needing reorder**: Products where total quantity_on_hand across all warehouses is less than or equal to the product's reorder_point

**Active products only**: Exclude products where discontinued = true unless specifically analyzing discontinued items

**Stock valuation**: quantity_on_hand * unit_cost for each product

## Example Query

-- Find products below reorder point across all warehouses
SELECT
    p.product_id,
    p.product_name,
    p.reorder_point,
    SUM(i.quantity_on_hand) as total_stock,
    p.unit_cost,
    (p.reorder_point - SUM(i.quantity_on_hand)) as units_to_reorder
FROM products p
JOIN inventory i ON p.product_id = i.product_id
WHERE p.discontinued = false
GROUP BY p.product_id, p.product_name, p.reorder_point, p.unit_cost
HAVING SUM(i.quantity_on_hand) <= p.reorder_point
ORDER BY units_to_reorder DESC;
""",
    },
]

# Create skill loading tool
@tool
def load_skill(skill_name: str) -> str:
    """Load the full content of a skill into the agent's context.

    Use this when you need detailed information about how to handle a specific
    type of request. This will provide you with comprehensive instructions,
    policies, and guidelines for the skill area.

    Args:
        skill_name: The name of the skill to load (e.g., "sales_analytics", "inventory_management")
    """
    # Find and return the requested skill
    for skill in SKILLS:
        if skill["name"] == skill_name:
            return f"Loaded skill: {skill_name}\n\n{skill['content']}"

    # Skill not found
    available = ", ".join(s["name"] for s in SKILLS)
    return f"Skill '{skill_name}' not found. Available skills: {available}"

# Create skill middleware
class SkillMiddleware(AgentMiddleware):
    """Middleware that injects skill descriptions into the system prompt."""

    # Register the load_skill tool as a class variable
    tools = [load_skill]

    def __init__(self):
        """Initialize and generate the skills prompt from SKILLS."""
        # Build skills prompt from the SKILLS list
        skills_list = []
        for skill in SKILLS:
            skills_list.append(
                f"- **{skill['name']}**: {skill['description']}"
            )
        self.skills_prompt = "\n".join(skills_list)

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        """Sync: Inject skill descriptions into system prompt."""
        # Build the skills addendum
        skills_addendum = (
            f"\n\n## Available Skills\n\n{self.skills_prompt}\n\n"
            "Use the load_skill tool when you need detailed information "
            "about handling a specific type of request."
        )

        # Append to system message content blocks
        new_content = list(request.system_message.content_blocks) + [
            {"type": "text", "text": skills_addendum}
        ]
        new_system_message = SystemMessage(content=new_content)
        modified_request = request.override(system_message=new_system_message)
        return handler(modified_request)

# Initialize your chat model (replace with your model)
# Example: from langchain_anthropic import ChatAnthropic
# model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-5.5")

# Create the agent with skill support
agent = create_agent(
    model,
    system_prompt=(
        "You are a SQL query assistant that helps users "
        "write queries against business databases."
    ),
    middleware=[SkillMiddleware()],
    checkpointer=InMemorySaver(),
)

# Example usage
if __name__ == "__main__":
    # Configuration for this conversation thread
    thread_id = str(uuid7())
    config = {"configurable": {"thread_id": thread_id}}

    # Ask for a SQL query
    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": (
                        "Write a SQL query to find all customers "
                        "who made orders over $1000 in the last month"
                    ),
                }
            ]
        },
        config
    )

    # Print the conversation
    for message in result["messages"]:
        if hasattr(message, 'pretty_print'):
            message.pretty_print()
        else:
            print(f"{message.type}: {message.content}")

This complete example includes:

Skill definitions with full database schemas
The load_skill tool for on-demand loading
SkillMiddleware that injects skill descriptions into the system prompt
Agent creation with middleware and checkpointer
Example usage showing how the agent loads skills and writes SQL queries

To run this, you'll need to:

Install required packages: pip install langchain langchain-openai langgraph
Set your API key (e.g., export OPENAI_API_KEY=...)
Replace the model initialization with your preferred LLM provider

Implementation variations

This tutorial implemented skills as in-memory Python dictionaries loaded through tool calls. However, there are several ways to implement progressive disclosure with skills:

Storage backends:

In-memory (this tutorial): Skills defined as Python data structures, fast access, no I/O overhead
File system (Claude Code approach): Skills as directories with files, discovered via file operations like read_file
Remote storage: Skills in S3, databases, Notion, or APIs, fetched on-demand

Skill discovery (how the agent learns which skills exist):

System prompt listing: Skill descriptions in system prompt (used in this tutorial)
File-based: Discover skills by scanning directories (Claude Code approach)
Registry-based: Query a skill registry service or API for available skills
Dynamic lookup: List available skills via a tool call

Progressive disclosure strategies (how skill content is loaded):

Single load: Load entire skill content in one tool call (used in this tutorial)
Paginated: Load skill content in multiple pages/chunks for large skills
Search-based: Search within a specific skill's content for relevant sections (e.g., using grep/read operations on skill files)
Hierarchical: Load skill overview first, then drill into specific subsections

Size considerations (uncalibrated mental model - optimize for your system):

Small skills (\< 1K tokens / \~750 words): Can be included directly in system prompt and cached with prompt caching for cost savings and faster responses
Medium skills (1-10K tokens / \~750-7.5K words): Benefit from on-demand loading to avoid context overhead (this tutorial)
Large skills (> 10K tokens / \~7.5K words, or > 5-10% of context window): Should use progressive disclosure techniques like pagination, search-based loading, or hierarchical exploration to avoid consuming excessive context

The choice depends on your requirements: in-memory is fastest but requires redeployment for skill updates, while file-based or remote storage enables dynamic skill management without code changes.

Progressive disclosure and context engineering

Progressive disclosure is fundamentally a context engineering technique - you're managing what information is available to the agent and when. This tutorial focused on loading database schemas, but the same principles apply to other types of context.

### Combining with few-shot prompting

For the SQL query use case, you could extend progressive disclosure to dynamically load few-shot examples that match the user's query:

Example approach:

User asks: "Find customers who haven't ordered in 6 months"
Agent loads sales_analytics schema (as shown in this tutorial)
Agent also loads 2-3 relevant example queries (via semantic search or tag-based lookup):
- Query for finding inactive customers
- Query with date-based filtering
- Query joining customers and orders tables
Agent writes query using both schema knowledge AND example patterns

This combination of progressive disclosure (loading schemas on-demand) and dynamic few-shot prompting (loading relevant examples) creates a powerful context engineering pattern that scales to large knowledge bases while providing high-quality, grounded outputs.

Next steps

Learn about middleware for more dynamic agent behaviors
Explore context engineering techniques for managing agent context
Explore the handoffs pattern for sequential workflows
Read the subagents pattern for parallel task routing
See multi-agent patterns for other approaches to specialized agents
Use LangSmith to debug and monitor skill loading

<div> <Callout icon="terminal-2"> Connect these docs to your agent of choice via MCP for real-time answers. </Callout>

<Callout icon="edit"> Edit this page on GitHub or file an issue. </Callout> </div>

Agent 与工具（2/3）