vs-mcp-agent

MCP server framework for Viveka Sutra — build agents that expose tools, resources, and prompts over the Model Context Protocol, with pluggable server implementations, lifecycle hooks, and unified auth.

---

Overview

vs-mcp-agent provides everything needed to build an MCP server: base classes for tools, resources, and prompts; decorators that self-register them at import time; a pluggable server factory; and lifecycle hooks. VsFastMcpServer ships as the built-in implementation backed by FastMCP, and auto-registers as "fastmcp" when its module is imported.

The library is server-agnostic by design. VsMcpServerFactory maps string keys to VsMcpServer subclasses — swap or add implementations without changing application code.

---

The Problem It Solves

Every MCP agent needs tools, auth, and a server. Without a shared framework, each agent re-implements the same wiring boilerplate.

Without vs-mcp-agent

from fastmcp import FastMCP

mcp = FastMCP(name="my-agent")

# manually wire each tool
@mcp.tool(name="search_docs")
async def search_docs(query: str) -> dict:
    # auth? write it yourself
    # error handling? write it yourself
    results = await _do_search(query)
    return {"results": results}

mcp.run(transport="streamable-http")

With vs-mcp-agent

from vs_mcp_agent.decorator.tool import tool
from vs_mcp_agent.schema.vs_tool_response import VsToolResponse
from vs_mcp_agent.auth.vs_mcp_security import VsMcpSecurity

@tool(name="search_docs", description="Search VS library documentation",
      guards=[VsMcpSecurity(roles=["user"])])
async def search_docs(query: str) -> VsToolResponse:
    return VsToolResponse(status="success", result=await _do_search(query))

Tools self-register, auth is declarative, and the server wires everything at startup.

---

Installation

pip install vs-mcp-agent

With FastMCP server support:

pip install vs-mcp-agent[fastmcp]

With auth support:

pip install vs-mcp-agent[fastmcp,security]

---

Dependencies

Library	Required	Purpose
pydantic	Yes	Tool, resource, and prompt input/output schemas
vs-common	Yes	Config, logging
fastmcp	No — install with [fastmcp] extra	FastMCP server implementation
uvicorn	No — install with [fastmcp] extra	ASGI server for HTTP transports
vs-security	No — install with [security] extra	JWT auth for VsMcpAuthMiddleware and VsMcpSecurity

---

Configuration

All MCP config keys live under the [agent] section of config.ini.

Key	Default	Description
agent.name	vs-agent	MCP server name
agent.version	0.1.0	MCP server version
agent.description	""	MCP server description
agent.transport	streamable-http	Transport: stdio, sse, streamable-http
agent.host	0.0.0.0	Host to bind (HTTP transports only)
agent.port	8080	Port to bind (HTTP transports only)
agent.enable_auth	false	Enable request authentication middleware (HTTP transports only)

config.ini example:

[agent]
name = my-agent
version = 1.0.0
transport = streamable-http
host = 0.0.0.0
port = 8080
enable_auth = true

[auth]
secret_key = your-secret-key
algorithm = HS256

[logging]
level = INFO
file_path = ./logs/agent.log

Transport options:

Transport	Description
stdio	Standard input/output — for local use by Claude Desktop and similar clients
sse	HTTP + Server-Sent Events
streamable-http	HTTP with streamable responses — recommended for network-accessible agents

host and port are ignored for stdio transport.

---

Quick Start

from vs_common.config.vs_ini_config import VsIniConfig
from vs_common.log.vs_log_manager import VsLogManager
from vs_common.schema.vs_log_config import VsLogConfig
from vs_mcp_agent.server.vs_fast_mcp_server import VsFastMcpServer  # noqa — auto-registers "fastmcp"
from vs_mcp_agent.factory.vs_mcp_server_factory import VsMcpServerFactory
from vs_mcp_agent.decorator.vs_mcp_server_registry import mcp_server_registry

import my_agent.tools    # noqa — registers @tool functions
import my_agent.resources  # noqa — registers @resource classes
import my_agent.prompts    # noqa — registers @prompt classes


@mcp_server_registry(hooks="my_agent.lifecycle")
def main():
    config = VsIniConfig("config.ini")
    VsLogManager.init(VsLogConfig(level=config.get("logging.level", default="INFO")))

    server = VsMcpServerFactory.get("fastmcp", config)
    server.run()


if __name__ == "__main__":
    main()

---

How It All Fits Together

Application Startup
    └── import VsFastMcpServer         # auto-registers "fastmcp" into VsMcpServerFactory
    └── import my_agent.tools          # @tool decorators self-register into VsToolRegistry
    └── import my_agent.resources      # @resource decorators self-register into VsResourceRegistry
    └── import my_agent.prompts        # @prompt decorators self-register into VsPromptRegistry

@mcp_server_registry(hooks="my_agent.lifecycle")
    └── imports my_agent.lifecycle     # @mcp_startup / @mcp_shutdown self-register into hook lists

VsMcpServerFactory.get("fastmcp", config)
    └── constructs VsAgentMcpConfig(config)
    └── instantiates VsFastMcpServer(mcp_config)

server.run()
    ├── calls startup() → runs @mcp_startup hooks in order
    ├── wires VsToolRegistry → FastMCP tools
    ├── wires VsResourceRegistry → FastMCP resources
    ├── wires VsPromptRegistry → FastMCP prompts
    ├── wires built-in tools: ping, health
    └── starts server on configured transport

---

VsMcpServer

Abstract base class for all MCP server implementations. Provides lifecycle hook execution and logging. All concrete servers must extend this.

from vs_mcp_agent.server.vs_mcp_server import VsMcpServer
from vs_mcp_agent.config.vs_agent_mcp_config import VsAgentMcpConfig
from vs_mcp_agent.decorator.vs_mcp_server_decorator import mcp_server
from vs_mcp_agent.factory.vs_mcp_server_factory import VsMcpServerFactory


@mcp_server("my-mcp")
class MyCustomMcpServer(VsMcpServer):

    def __init__(self, config: VsAgentMcpConfig):
        super().__init__(config)

    def run(self) -> None:
        ...  # start your MCP server here


VsMcpServerFactory.register("my-mcp", MyCustomMcpServer)

startup() and shutdown() on the base class iterate @mcp_startup and @mcp_shutdown hooks. Always call super().startup() and super().shutdown() to keep lifecycle hooks running.

---

VsMcpServerFactory

Registry that maps string keys to VsMcpServer subclasses. Used to resolve and instantiate a server at startup without hardcoding the implementation.

from vs_mcp_agent.factory.vs_mcp_server_factory import VsMcpServerFactory

# register a custom implementation
VsMcpServerFactory.register("my-mcp", MyCustomMcpServer)

# get an instance — factory constructs VsAgentMcpConfig(config) internally
server = VsMcpServerFactory.get("my-mcp", config)
server.run()

VsFastMcpServer registers itself as "fastmcp" automatically when its module is imported — no manual register() call needed.

from vs_mcp_agent.server.vs_fast_mcp_server import VsFastMcpServer  # noqa — triggers auto-register

server = VsMcpServerFactory.get("fastmcp", config)

---

@mcp_server

Decorator that marks a class as a named MCP server implementation. Validates that the class extends VsMcpServer.

from vs_mcp_agent.decorator.vs_mcp_server_decorator import mcp_server

@mcp_server("my-mcp")
class MyCustomMcpServer(VsMcpServer):
    ...

Sets cls._vs_mcp_server_key = key on the class. Used by mcp_server_registry when scanning packages.

---

mcp_server_registry

Scans packages for @mcp_server-decorated classes and @mcp_startup/@mcp_shutdown hooks, and auto-registers them. Works as a decorator on the main() function or as a plain call. All parameters are optional.

from vs_mcp_agent.decorator.vs_mcp_server_registry import mcp_server_registry

# as a decorator
@mcp_server_registry(server="my_pkg.server", hooks="my_pkg.lifecycle")
def main():
    ...

# as a plain call
mcp_server_registry(hooks="my_agent.lifecycle")

Parameter	Description
server	Package path to scan for @mcp_server-decorated classes. Use for custom server implementations. Not needed for "fastmcp" — it auto-registers on import.
hooks	Package path to scan for @mcp_startup / @mcp_shutdown functions. Importing the module triggers self-registration.

---

Lifecycle Hooks

Use @mcp_startup and @mcp_shutdown to run code when the MCP server starts and stops — loading tool indexes, closing connections, etc.

from vs_mcp_agent.lifecycle.vs_mcp_lifecycle import mcp_startup, mcp_shutdown

@mcp_startup
async def load_tool_index():
    await ToolIndex.load()

@mcp_shutdown
async def close_connections():
    await ToolIndex.close()

Hooks self-register into global lists at import time. They are executed in registration order. Both sync and async hooks are supported.

Register hook modules with mcp_server_registry so they are imported before the server starts:

@mcp_server_registry(hooks="my_agent.lifecycle")
def main():
    ...

VsMcpServer.startup() iterates _mcp_startup_hooks. VsMcpServer.shutdown() iterates _mcp_shutdown_hooks.

---

Tools

Tools are actions the agent can perform. The MCP client (orchestrator or Claude) calls tools to delegate work.

Function-style (stateless)

from vs_mcp_agent.decorator.tool import tool
from vs_mcp_agent.schema.vs_tool_response import VsToolResponse

@tool(name="search_docs", description="Search VS library documentation")
async def search_docs(query: str) -> VsToolResponse:
    results = await _do_search(query)
    return VsToolResponse(status="success", result=results, summary="Found results")

Class-style (stateful, with dependency injection)

Use when the tool needs a client, config, or any injected dependency.

from pydantic import BaseModel
from vs_mcp_agent.base.vs_base_tool import VsBaseTool
from vs_mcp_agent.decorator.tool import tool
from vs_mcp_agent.schema.vs_tool_response import VsToolResponse

@tool(name="search_docs", description="Search VS library documentation")
class SearchDocsTool(VsBaseTool):

    class Input(BaseModel):
        query: str

    def __init__(self):
        self._client = MySearchClient()

    async def execute(self, input: Input) -> VsToolResponse:
        results = await self._client.search(input.query)
        return VsToolResponse(status="success", result=results, summary="Found results")

Both styles self-register in VsToolRegistry at import time. VsFastMcpServer reads VsToolRegistry at run() and wires everything into FastMCP automatically.

@tool parameters

Parameter	Type	Required	Default	Description
name	str	Yes	—	Tool name exposed to the MCP client
description	str	Yes	—	Shown in MCP tool list
roles	List[str]	No	[]	Allowed roles (informational — enforce via guards)
guards	List[Callable]	No	[]	Async callables run before the tool executes

VsToolResponse

All tool functions must return VsToolResponse.

Field	Type	Description
status	"success" | "error" | "clarification" | "not_supported"	Outcome signal for the orchestrator
result	Optional[Any]	Return payload on success
summary	str	Human-readable summary of what happened
confidence	Optional[float]	Confidence score (0–1)
artifacts_extracted	List[ArtifactEntry]	Structured data extracted during execution
user_preferences_extracted	List[UserPreferenceEntry]	User preferences inferred during execution
active_highlights	List[ActiveHighlight]	Key highlights from the result
tokens_consumed	int	Token count if the tool called an LLM
error	Optional[ErrorDetail]	Error detail when status="error"
clarification_needed	Optional[ClarificationDetail]	Clarification questions when status="clarification"
not_supported	Optional[NotSupportedDetail]	Reason when status="not_supported"

Status values:

Status	When to use
success	Tool completed successfully
error	Tool failed — populate error with ErrorDetail
clarification	Need more input from the user — populate clarification_needed
not_supported	Request is out of scope — populate not_supported with a reason

---

Resources

Resources expose read-only data — documents, schemas, configs — that the MCP client can fetch by URI.

from pydantic import BaseModel
from vs_mcp_agent.base.vs_base_resource import VsBaseResource
from vs_mcp_agent.decorator.resource import resource
from vs_mcp_agent.schema.vs_resource_response import VsResourceResponse

@resource(uri="docs://api-schema", description="Returns the API schema document")
class ApiSchemaResource(VsBaseResource):

    class Input(BaseModel):
        version: str = "latest"

    async def fetch(self, input: Input) -> VsResourceResponse:
        schema = await _load_schema(input.version)
        return VsResourceResponse(status="success", result=schema)

@resource parameters:

Parameter	Type	Required	Description
uri	str	Yes	URI the MCP client uses to fetch this resource
description	str	Yes	Shown in MCP resource list

VsResourceResponse fields:

Field	Type	Description
status	"success" | "error" | "not_supported"	Outcome
result	Optional[Any]	Resource data on success
error	Optional[ErrorDetail]	Error detail when status="error"
not_supported	Optional[NotSupportedDetail]	Reason when status="not_supported"

---

Prompts

Prompts expose reusable prompt templates that the MCP client can request by name.

from pydantic import BaseModel
from vs_mcp_agent.base.vs_base_prompt import VsBasePrompt
from vs_mcp_agent.decorator.prompt import prompt
from vs_mcp_agent.schema.vs_prompt_response import VsPromptResponse

@prompt(name="summarise_doc", description="Prompt to summarise a document")
class SummariseDocPrompt(VsBasePrompt):

    class Input(BaseModel):
        topic: str
        length: str = "short"

    async def generate(self, input: Input) -> VsPromptResponse:
        text = f"Summarise the following in {input.length} form. Topic: {input.topic}"
        return VsPromptResponse(status="success", prompt=text)

@prompt parameters:

Parameter	Type	Required	Description
name	str	Yes	Prompt name exposed to the MCP client
description	str	Yes	Shown in MCP prompt list

VsPromptResponse fields:

Field	Type	Description
status	"success" | "error"	Outcome
prompt	Optional[str]	The rendered prompt string on success
error	Optional[ErrorDetail]	Error detail when status="error"

---

Authorization

VsMcpSecurity

Guard for MCP-only tools. Reads the auth context set by VsMcpAuthMiddleware after request authentication.

from vs_mcp_agent.auth.vs_mcp_security import VsMcpSecurity

@tool(name="admin_tool", description="Admin only", guards=[VsMcpSecurity(roles=["admin"])])
async def admin_tool(query: str) -> VsToolResponse:
    ...

No roles (auth only):

guards=[VsMcpSecurity()]  # rejects unauthenticated callers, any role allowed

Multiple roles (any match):

guards=[VsMcpSecurity(roles=["admin", "editor"])]  # passes if caller has admin OR editor

Stdio authentication runs through the same authenticator before each MCP request. VsAuthRequest.transport is stdio, with empty headers and the MCP method name. For bearer sessions, explicitly set VsBearerTokenAuthenticator(session, stdio_token_env="VS_AGENT_TOKEN") and have the launcher supply that environment variable. Missing credentials reject dispatch. These credentials identify the launched process. Disable VS console logging before imports and log to a file so stdout contains only protocol messages. The former agent.auth_enabled configuration key is no longer read; migrate it to agent.enable_auth.

Enable auth middleware by setting agent.enable_auth = true in config.ini and passing an authenticator instance. VsFastMcpServer wires the HTTP and per-message context middleware automatically. The previous implicit JWT provider is replaced by an explicit instance (add stdio_token_env for stdio bearer authentication):

from vs_security import VsBearerTokenAuthenticator, VsJWTProvider

server = VsFastMcpServer(
    VsAgentMcpConfig(config),
    authenticator=VsBearerTokenAuthenticator(VsJWTProvider(config)),
)

Custom authenticators implement VsRequestAuthenticator.authenticate(VsAuthRequest) -> VsAuthContext from vs-security. They must support concurrent calls across transport threads and event loops. Missing instances fail construction when authentication is enabled; disabled authentication requires no security dependency.

Custom guards

Any async callable that raises PermissionError on failure works as a guard:

async def require_verified_user():
    ctx = get_auth_context()
    if not ctx or not ctx.is_verified:
        raise PermissionError("Verified account required")

@tool(name="sensitive_tool", description="Verified users only",
      guards=[VsMcpSecurity(roles=["user"]), require_verified_user])
async def sensitive_tool(query: str) -> VsToolResponse:
    ...

Multiple guards run in order — all must pass before the tool executes. Guarded tool functions and guards may be synchronous or asynchronous; synchronous callables run in a worker thread with the request context available, and exceptions propagate without executing later guards or the tool.

VsMcpAuthMiddleware

Starlette middleware that delegates every incoming request to the supplied authenticator (except /.well-known/ paths). Sets the auth context so VsMcpSecurity and get_auth_context() work inside tools.

VsFastMcpServer registers VsMcpAuthMiddleware automatically when agent.enable_auth = true. You do not need to wire it manually.

from vs_mcp_agent.auth.vs_mcp_auth import get_auth_context

@tool(name="whoami", description="Return caller identity")
async def whoami() -> VsToolResponse:
    ctx = get_auth_context()
    return VsToolResponse(status="success", result={"username": ctx.username})

---

Built-in Tools

Every VsFastMcpServer instance registers two tools automatically:

Tool	Description
ping	Liveness check. Returns {"pong": true, "timestamp": "..."}.
health	Returns agent name, version, transport, and uptime in seconds.

These are always available to MCP clients without any configuration.

---

Extending vs-mcp-agent

Adding a custom MCP server implementation

Implement VsMcpServer, decorate with @mcp_server, and register:

from vs_mcp_agent.server.vs_mcp_server import VsMcpServer
from vs_mcp_agent.config.vs_agent_mcp_config import VsAgentMcpConfig
from vs_mcp_agent.decorator.vs_mcp_server_decorator import mcp_server
from vs_mcp_agent.factory.vs_mcp_server_factory import VsMcpServerFactory


@mcp_server("my-mcp")
class MyMcpServer(VsMcpServer):

    def __init__(self, config: VsAgentMcpConfig):
        super().__init__(config)

    def run(self) -> None:
        self._wire_tools()
        # start your server

    def _wire_tools(self) -> None:
        from vs_mcp_agent.registry.vs_tool_registry import VsToolRegistry
        for name, entry in VsToolRegistry.get_all().items():
            ...  # register into your server


VsMcpServerFactory.register("my-mcp", MyMcpServer)

Then use it by key:

server = VsMcpServerFactory.get("my-mcp", config)
server.run()

Or scan the package automatically:

@mcp_server_registry(server="my_pkg.server")
def main():
    server = VsMcpServerFactory.get("my-mcp", config)
    server.run()

---

Error Handling

Exceptions raised inside tool, resource, or prompt functions are caught by VsFastMcpServer and returned as MCP error results — the server does not crash.

For expected failures, use VsToolResponse with status="error" rather than raising:

from vs_mcp_agent.schema.vs_tool_response import VsToolResponse, ErrorDetail

@tool(name="get_doc", description="Fetch a document by ID")
async def get_doc(doc_id: str) -> VsToolResponse:
    doc = await repo.find(doc_id)
    if doc is None:
        return VsToolResponse(
            status="error",
            error=ErrorDetail(code="NOT_FOUND", message=f"Document '{doc_id}' not found", retryable=False),
        )
    return VsToolResponse(status="success", result=doc)

---

Class Reference

---

VsMcpServer

Abstract base class for all MCP server implementations.

Constructor:

Parameter	Type	Description
config	VsAgentMcpConfig	Parsed MCP configuration

Methods:

Method	Signature	Description
startup	async startup() -> None	Runs @mcp_startup hooks in order. Call super().startup() to keep hook execution.
shutdown	async shutdown() -> None	Runs @mcp_shutdown hooks in order. Call super().shutdown() to keep hook execution.
run	run() -> None	Abstract. Start the MCP server.

---

VsMcpServerFactory

Registry mapping string keys to VsMcpServer subclasses. Thread-safe class-level dict.

Methods:

Method	Signature	Description
register	register(key: str, server_class: Type[VsMcpServer]) -> None	Register a server class. Raises TypeError if not a VsMcpServer subclass.
get	get(key: str, config: VsBaseConfig) -> VsMcpServer	Construct and return a server instance. Raises KeyError if key not registered. Wraps config in VsAgentMcpConfig internally.

Notes:

• VsFastMcpServer auto-registers as "fastmcp" when its module is imported.
• Import VsFastMcpServer before calling VsMcpServerFactory.get("fastmcp", config).

---

VsAgentMcpConfig

Parsed MCP configuration. Constructed from VsBaseConfig by reading the [agent] section.

Fields:

Field	Type	Default	Description
name	str	vs-agent	MCP server name
version	str	0.1.0	MCP server version
description	str	""	MCP server description
transport	str	streamable-http	Transport type
host	str	0.0.0.0	Bind host
port	int	8080	Bind port
auth_enabled	bool	False	Enable request authentication middleware

Notes:

• Raises ValueError on construction if transport is not one of stdio, sse, streamable-http.

---

VsFastMcpServer

Built-in VsMcpServer implementation backed by FastMCP. Auto-registers as "fastmcp" on import.

Constructor: VsFastMcpServer(config: VsAgentMcpConfig, authenticator=None). An authenticator is required when config.auth_enabled is true. VsMcpServerFactory.get(key, config, authenticator=...) forwards the instance to enabled servers.

Behaviour at run():

1. Reads VsToolRegistry, VsResourceRegistry, VsPromptRegistry and wires all entries into FastMCP.
2. Applies guard proxies to guarded tools.
3. Registers built-in ping and health tools.
4. If auth_enabled, wires VsMcpAuthMiddleware.
5. For stdio transport: calls mcp.run(transport="stdio").
6. For HTTP transports: wraps the ASGI lifespan to call startup()/shutdown(), then starts uvicorn.

Notes:

• Requires fastmcp and uvicorn — install with pip install vs-mcp-agent[fastmcp].
• mcp property exposes the underlying FastMCP instance for advanced configuration.

---

@mcp_server

Decorator. Marks a class as a named MCP server implementation. Validates that the class extends VsMcpServer.

Parameters:

Parameter	Type	Required	Description
key	str	Yes	Registry key used with VsMcpServerFactory.get(key, config)

---

mcp_server_registry

Scans packages for @mcp_server classes and @mcp_startup/@mcp_shutdown functions and registers them. Works as a function decorator or a plain call.

Parameters:

Parameter	Type	Description
server	Optional[str]	Package path to scan for @mcp_server classes
hooks	Optional[str]	Package path to scan for lifecycle hook functions

---

@mcp_startup / @mcp_shutdown

Decorators. Register a function into the global _mcp_startup_hooks or _mcp_shutdown_hooks list at import time. Both sync and async functions are supported.

from vs_mcp_agent.lifecycle.vs_mcp_lifecycle import mcp_startup, mcp_shutdown

@mcp_startup
async def on_start():
    ...

@mcp_shutdown
def on_stop():
    ...

Hooks are called in registration order by VsMcpServer.startup() and VsMcpServer.shutdown().

---

@tool

Decorator. Registers a function or class in VsToolRegistry at import time.

Parameters:

Parameter	Type	Required	Default	Description
name	str	Yes	—	Tool name exposed to the MCP client
description	str	Yes	—	Shown in MCP tool list
roles	List[str]	No	[]	Allowed roles (informational)
guards	List[Callable]	No	[]	Async callables run before the tool executes

Notes:

• Each name must be unique. Duplicate names raise ValueError.
• Class-style: target must extend VsBaseTool and implement execute(input) -> VsToolResponse.
• Function-style: target must be an async callable.

---

VsBaseTool

Abstract base class for class-style tools.

Inner class:

Name	Description
Input	Pydantic BaseModel subclass. Define fields here for typed input.

Methods:

Method	Signature	Description
execute	async execute(input: Input) -> VsToolResponse	Abstract. Implement tool logic here.

---

@resource

Decorator. Registers a class in VsResourceRegistry at import time.

Parameters:

Parameter	Type	Required	Description
uri	str	Yes	URI the MCP client uses to fetch this resource
description	str	Yes	Shown in MCP resource list

Notes:

• Target must extend VsBaseResource and implement fetch(input) -> VsResourceResponse.
• Each uri must be unique. Duplicate URIs raise ValueError.

---

VsBaseResource

Abstract base class for resources.

Inner class:

Name	Description
Input	Pydantic BaseModel subclass. Define fields for fetch parameters.

Methods:

Method	Signature	Description
fetch	async fetch(input: Input) -> VsResourceResponse	Abstract. Return the resource data.

---

@prompt

Decorator. Registers a class in VsPromptRegistry at import time.

Parameters:

Parameter	Type	Required	Description
name	str	Yes	Prompt name exposed to the MCP client
description	str	Yes	Shown in MCP prompt list

Notes:

• Target must extend VsBasePrompt and implement generate(input) -> VsPromptResponse.
• Each name must be unique. Duplicate names raise ValueError.

---

VsBasePrompt

Abstract base class for prompts.

Inner class:

Name	Description
Input	Pydantic BaseModel subclass. Define fields for prompt parameters.

Methods:

Method	Signature	Description
generate	async generate(input: Input) -> VsPromptResponse	Abstract. Return the rendered prompt.

---

VsMcpSecurity

Guard for MCP-only tools. Reads the auth context set by VsMcpAuthMiddleware.

Constructor:

Parameter	Type	Default	Description
roles	Optional[List[str]]	[]	Required roles. Empty = any authenticated caller.

Behaviour:

Condition	Result
No auth context	Raises PermissionError("Unauthenticated request")
Auth context present, no roles required	Passes
Auth context present, caller has required role	Passes
Auth context present, caller lacks required role	Raises PermissionError

---

VsMcpAuthMiddleware

Starlette middleware. Calls the supplied authenticator on each request and scopes the shared auth context.

Constructor:

Parameter	Type	Description
app	ASGI app	The wrapped application
authenticator	VsRequestAuthenticator	Shared request authenticator from vs-security; replaces the former jwt_provider argument

Notes:

• VsFastMcpServer registers this automatically when agent.enable_auth = true.
• Paths starting with /.well-known/ are whitelisted and bypass auth.
• Uses the single shared VS context. Per-message middleware rebinds the identity for each MCP message, including requests within existing sessions. Context is reset in finally after success, failure, or cancellation.

---

VsToolRegistry

Class-level registry of all @tool-registered functions and classes. Thread-safe.

Methods:

Method	Signature	Description
register_class	register_class(name, description, tool_class, roles, guards) -> None	Register a class-style tool. Raises TypeError if not a VsBaseTool subclass.
register_fn	register_fn(name, description, fn, roles, guards) -> None	Register a function-style tool.
get_all	get_all() -> Dict[str, _ToolEntry]	Returns a snapshot of all registered tools.

---

VsResourceRegistry

Class-level registry of all @resource-registered classes. Thread-safe.

Methods:

Method	Signature	Description
register	register(uri, description, resource_class) -> None	Register a resource class. Raises TypeError if not a VsBaseResource subclass.
get_all	get_all() -> Dict[str, _ResourceEntry]	Returns a snapshot of all registered resources.

---

VsPromptRegistry

Class-level registry of all @prompt-registered classes. Thread-safe.

Methods:

Method	Signature	Description
register	register(name, description, prompt_class) -> None	Register a prompt class. Raises TypeError if not a VsBasePrompt subclass.
get_all	get_all() -> Dict[str, _PromptEntry]	Returns a snapshot of all registered prompts.