Capstone: Task Manager, Server and Client

What this is

No new concepts in this lesson. This is the capstone: a complete Task Manager server plus a Gemini client, combining every piece from Lessons 2 through 25. If you can read the code and explain why every tool, the lifespan state, the transport switch, and the persistent session are there, you've built a full working MCP server and client, from the ground up.

What it does

A Task Manager server (add_task, complete_task, list_tasks, a tasks://{id} resource, a summarize_tasks prompt), runnable over stdio or Streamable HTTP with the exact same code, plus a Gemini client that connects to it, binds its tools, and runs a real multi-turn conversation, adding tasks, completing one, and asking the model to summarize where things stand.

Where each piece came from

@dataclass
class TaskManagerState:
    tasks: dict[int, dict] = field(default_factory=dict)
    next_id: int = 1


@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[TaskManagerState]:
    logger.info("startup")
    state = TaskManagerState()
    try:
        yield state
    finally:
        logger.info("shutdown")


mcp = FastMCP("task-manager-server", lifespan=lifespan)

Lesson 18: tasks live in a proper TaskManagerState, set up and torn down around the server's lifetime, not a bare module-level dict like the beginner checkpoint used.

@mcp.tool()
def complete_task(ctx: Context, task_id: int) -> str:
    """Mark a task as complete."""
    state: TaskManagerState = ctx.request_context.lifespan_context
    if task_id not in state.tasks:
        raise ValueError(f"No task with id {task_id}.")
    state.tasks[task_id]["done"] = True
    return f"Completed task {task_id}."

Lesson 7: completing a task that doesn't exist raises a clear error, isError=True, the server keeps running. Lesson 8: every log line goes through logging, never print().

import argparse

parser = argparse.ArgumentParser()
parser.add_argument("--transport", default="stdio", choices=["stdio", "streamable-http"])
args = parser.parse_args()

mcp.run(transport=args.transport)

Lesson 20: server.py runs over stdio by default, or over Streamable HTTP with --transport streamable-http, same tools, resources, and prompt either way, only the transport argument differs.

async with client.session("tasks") as session:
    tools = await load_mcp_tools(session)
    tools_by_name = {tool.name: tool for tool in tools}

    model = ChatGoogleGenerativeAI(model="gemini-3.5-flash-lite").bind_tools(tools)
    messages: list = []

    for question in [
        "Add a task to write the MCP course.",
        "Add a task to review the pull request.",
        "Mark the first task as complete.",
        "Summarize where things stand across all my tasks.",
    ]:
        answer = await ask(model, tools_by_name, messages, question)

Lessons 15-16: langchain-mcp-adapters and the full ask/call/respond loop, running a real multi-turn conversation. Lesson 19's lesson resolved: like the intermediate checkpoint, this holds one persistent session open for the whole conversation (client.session("tasks") + load_mcp_tools(session)), rather than the ephemeral, one-session-per-call client.get_tools() from Lessons 15-17. A stateless calculator doesn't care either way, but the Task Manager's lifespan state (Lesson 18) would reset every call under the ephemeral pattern, a second add_task would get id 1 again instead of 2.

Where to go from here

This capstone is deliberately still small enough to read start to finish in one sitting. A production version of this same server would add: persistent storage instead of in-memory state, real authentication if deployed over HTTP (Lesson 21), and tracing to see what the model actually did across a session. Every piece those additions would build on, tools, resources, prompts, transports, error handling, security, you've now built yourself, from scratch, in this course.