Production AI systems, designed for AWS.

import os
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langgraph.errors import GraphRecursionError
from langchain_aws import ChatBedrockConverse
from langchain_mcp_adapters.client import MultiServerMCPClient

# Full step-by-step tracing + eval in LangSmith (zero code changes downstream)
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "agentic-platform-prod"

llm = ChatBedrockConverse(model="anthropic.claude-sonnet-4-20250514-v1:0", temperature=0)

# Only these tools may ever be bound — everything else is ignored
ALLOWED = {"list_visits", "get_client", "check_availability", "schedule_visit"}
WRITE_TOOLS = {"schedule_visit"}        # require human approval before executing

async def build_agent():
    mcp = MultiServerMCPClient({
        "scheduling": {"url": "https://mcp.internal/scheduling", "transport": "streamable_http"},
        "records":    {"url": "https://mcp.internal/records",    "transport": "streamable_http"},
    })
    tools = [t for t in await mcp.get_tools() if t.name in ALLOWED]   # allow-list

    return create_react_agent(
        llm, tools,
        checkpointer=MemorySaver(),                # resumable state for approvals
        interrupt_before=["tools"],                # pause before any tool call
        prompt="You are an operations assistant. Never invent data; cite tool results.",
    )

async def run(agent, message: str, thread_id: str, approve):
    cfg = {"configurable": {"thread_id": thread_id}, "recursion_limit": 12}  # loop cap
    try:
        state = await agent.ainvoke({"messages": [("user", message)]}, cfg)

        # Paused on a pending tool call — gate writes through a human
        while (pending := state.get("__interrupt__")):
            call = state["messages"][-1].tool_calls[0]
            if call["name"] in WRITE_TOOLS and not await approve(call):
                return "Action declined by reviewer."
            state = await agent.ainvoke(None, cfg)   # resume the graph

        return state["messages"][-1].content
    except GraphRecursionError:
        return "Stopped: step budget exceeded before reaching an answer."

import boto3

agent = boto3.client("bedrock-agent-runtime", region_name="us-east-1")
MODEL = "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-sonnet-4-20250514-v1:0"

class NoGroundedAnswer(Exception): ...

def answer(question: str, kb_id: str, user_groups: list[str]) -> dict:
    resp = agent.retrieve_and_generate(
        input={"text": question},
        retrieveAndGenerateConfiguration={
            "type": "KNOWLEDGE_BASE",
            "knowledgeBaseConfiguration": {
                "knowledgeBaseId": kb_id,
                "modelArn": MODEL,
                "retrievalConfiguration": {
                    "vectorSearchConfiguration": {
                        "numberOfResults": 5,
                        "overrideSearchType": "HYBRID",        # semantic + keyword
                        # Doc-level access control — user only retrieves permitted docs
                        "filter": {"in": {"key": "acl_group", "value": user_groups}},
                    },
                },
                "generationConfiguration": {
                    "guardrailConfiguration": {
                        "guardrailId": "kb-guardrail", "guardrailVersion": "1",
                    },
                },
            },
        },
    )

    # Refuse-on-no-source: never let the model answer ungrounded
    cites = [c for c in resp["citations"] if c["retrievedReferences"]]
    if not cites:
        raise NoGroundedAnswer("no permitted sources matched the question")

    sources = [
        {
            "uri": r["location"]["s3Location"]["uri"],
            "score": r.get("metadata", {}).get("score"),
        }
        for c in cites for r in c["retrievedReferences"]
    ]
    return {"answer": resp["output"]["text"], "sources": sources}

from mcp.server.fastmcp import FastMCP, Context
import boto3

mcp = FastMCP("scheduling")
visits = boto3.resource("dynamodb").Table("visits")

class NotAuthorised(Exception): ...

def _require(ctx: Context, scope: str):
    # Scopes come from the Cognito token the client presented at connect
    if scope not in ctx.session.scopes:
        raise NotAuthorised(f"missing scope: {scope}")

# READ tool — typed signature IS the input schema the client validates
@mcp.tool()
def check_availability(staff_id: str, date: str, ctx: Context) -> list[str]:
    """Return open time slots for a staff member on YYYY-MM-DD."""
    _require(ctx, "scheduling:read")
    rows = visits.query(KeyConditionExpression=Key("staff").eq(staff_id)).get("Items", [])
    booked = {r["slot"] for r in rows if r["date"] == date}
    return [s for s in SLOTS if s not in booked]

# WRITE tool — separate scope; host gates it behind human approval
@mcp.tool()
def schedule_visit(staff_id: str, client_id: str, slot: str, ctx: Context) -> dict:
    """Book a visit. Side-effecting — requires the scheduling:write scope."""
    _require(ctx, "scheduling:write")
    visits.put_item(
        Item={"staff": staff_id, "client": client_id, "slot": slot},
        ConditionExpression="attribute_not_exists(slot)",   # idempotent
    )
    return {"status": "booked", "staff": staff_id, "slot": slot}

if __name__ == "__main__":
    mcp.run(transport="streamable-http")     # fronted by API Gateway + PrivateLink

from langchain_aws import ChatBedrock
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

class Triage(BaseModel):
    summary: str = Field(description="one-line summary")
    priority: str = Field(pattern="^(low|medium|high)$")
    next_action: str

parser = PydanticOutputParser(pydantic_object=Triage)

prompt = ChatPromptTemplate.from_template(
    "Triage this ticket.\n{format}\n\nContext:\n{context}\n\nTicket: {question}"
).partial(format=parser.get_format_instructions())

# Primary model, with an automatic fallback to a cheaper one on error
llm = ChatBedrock(model_id="anthropic.claude-sonnet-4-20250514-v1:0",
                 model_kwargs={"temperature": 0}).with_fallbacks(
    [ChatBedrock(model_id="anthropic.claude-3-5-haiku-20241022-v1:0")]
)

# LCEL: fetch context + question in parallel, then prompt → llm → parse
chain = (
    RunnableParallel(context=retriever, question=RunnablePassthrough())
    | prompt | llm | parser
)

# One definition — sync, async, streaming and batch all come for free
result: Triage = chain.invoke("Printer on floor 3 is jammed again")
batch = chain.batch(open_tickets, config={"max_concurrency": 8})