Skip to content

MCP

Bases: BoostedMagic

Source code in pydantic_ai_kernel/magics/mcp_magic.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class MCPMagic(BoostedMagic):
    @boosted_option(
        "--url",
        "-u",
        default=None,
        help="In case of streamable-http or sse transport, the url of the MCP server.",
    )
    @boosted_option(
        "--stdio_cmd",
        default=None,
        help="In case of stdio transport, the command of stdio server (ex: `python`)",
    )
    @boosted_option(
        "--stdio_args",
        default=None,
        help="In case of stdio transport, the args of the server (ex: `['mcp_server.py']`). Must be a list of str.",
    )
    @boosted_option(
        "--sse",
        action="store_true",
        default=False,
        help="Whether to use HTTP server with sse.",
    )
    @boosted_option(
        "--no-approval",
        default=False,
        action="store_true",
        help="If set, the MCP server will not need user approval for tool execution",
    )
    def line_mcp(
        self,
        url: str | None = None,
        stdio_cmd: str | None = None,
        stdio_args: list[str] | None = None,
        sse: bool = False,
        no_approval: bool = False,
    ):
        """
        %mcp : adds an MCP server to tools of the agent

        Examples :
        -------
            • %mcp --url http://127.0.0.1:8000/mcp : connects
                to server with streamable-http
            • %mcp --stdio_cmd python --stdio_args ['my_mcp.py'] :
                connects to MCP server through stdin / stdout
            • %mcp --url http://127.0.0.1:8000/sse : connects
                to server with sse
        """
        if url is not None:
            if sse:
                mcp_toolset = create_mcp_toolset("sse", url=url)
            else:
                mcp_toolset = create_mcp_toolset("streamable-http", url=url)
            to_display = f"located at : {url}"
        else:
            if stdio_cmd is None and stdio_args is None:
                raise MCPToolsetError(
                    "You must specify either an URL for http or sse connection, or --stdio_command and --stdio_args parameters for stdio connection."
                )
            mcp_toolset = create_mcp_toolset(
                "stdio", stdio_command=stdio_cmd, stdio_args=stdio_args
            )
            to_display = f"with stdio : {stdio_cmd}, {stdio_args}"
        self.kernel.log.info(f"Successfully connected to MCP server, {to_display}")
        self.kernel.Print(f"Successfully connected to MCP server, {to_display}")
        self.kernel: PydanticAIBaseKernel
        if self.kernel.toolsets is None:
            self.kernel.toolsets = [mcp_toolset]
        else:
            if no_approval:
                toolset = mcp_toolset
            else:
                toolset = ApprovalRequiredToolset(
                    mcp_toolset
                )  # by default, MCP requires
            # approval for all calls
            self.kernel.toolsets.append(toolset)

        # reset agent
        self.kernel._agent = self.kernel.create_agent()
        self.kernel.Print("Added MCP server as tool of the agent.")

line_mcp(url=None, stdio_cmd=None, stdio_args=None, sse=False, no_approval=False)

%mcp : adds an MCP server to tools of the agent

Examples :

• %mcp --url http://127.0.0.1:8000/mcp : connects
    to server with streamable-http
• %mcp --stdio_cmd python --stdio_args ['my_mcp.py'] :
    connects to MCP server through stdin / stdout
• %mcp --url http://127.0.0.1:8000/sse : connects
    to server with sse
Source code in pydantic_ai_kernel/magics/mcp_magic.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@boosted_option(
    "--url",
    "-u",
    default=None,
    help="In case of streamable-http or sse transport, the url of the MCP server.",
)
@boosted_option(
    "--stdio_cmd",
    default=None,
    help="In case of stdio transport, the command of stdio server (ex: `python`)",
)
@boosted_option(
    "--stdio_args",
    default=None,
    help="In case of stdio transport, the args of the server (ex: `['mcp_server.py']`). Must be a list of str.",
)
@boosted_option(
    "--sse",
    action="store_true",
    default=False,
    help="Whether to use HTTP server with sse.",
)
@boosted_option(
    "--no-approval",
    default=False,
    action="store_true",
    help="If set, the MCP server will not need user approval for tool execution",
)
def line_mcp(
    self,
    url: str | None = None,
    stdio_cmd: str | None = None,
    stdio_args: list[str] | None = None,
    sse: bool = False,
    no_approval: bool = False,
):
    """
    %mcp : adds an MCP server to tools of the agent

    Examples :
    -------
        • %mcp --url http://127.0.0.1:8000/mcp : connects
            to server with streamable-http
        • %mcp --stdio_cmd python --stdio_args ['my_mcp.py'] :
            connects to MCP server through stdin / stdout
        • %mcp --url http://127.0.0.1:8000/sse : connects
            to server with sse
    """
    if url is not None:
        if sse:
            mcp_toolset = create_mcp_toolset("sse", url=url)
        else:
            mcp_toolset = create_mcp_toolset("streamable-http", url=url)
        to_display = f"located at : {url}"
    else:
        if stdio_cmd is None and stdio_args is None:
            raise MCPToolsetError(
                "You must specify either an URL for http or sse connection, or --stdio_command and --stdio_args parameters for stdio connection."
            )
        mcp_toolset = create_mcp_toolset(
            "stdio", stdio_command=stdio_cmd, stdio_args=stdio_args
        )
        to_display = f"with stdio : {stdio_cmd}, {stdio_args}"
    self.kernel.log.info(f"Successfully connected to MCP server, {to_display}")
    self.kernel.Print(f"Successfully connected to MCP server, {to_display}")
    self.kernel: PydanticAIBaseKernel
    if self.kernel.toolsets is None:
        self.kernel.toolsets = [mcp_toolset]
    else:
        if no_approval:
            toolset = mcp_toolset
        else:
            toolset = ApprovalRequiredToolset(
                mcp_toolset
            )  # by default, MCP requires
        # approval for all calls
        self.kernel.toolsets.append(toolset)

    # reset agent
    self.kernel._agent = self.kernel.create_agent()
    self.kernel.Print("Added MCP server as tool of the agent.")