This is an automated email from the ASF dual-hosted git repository. hsluoyz pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/casbin-gateway.git
commit ae6633353c112bd236dd5172ccc69d3ae5387237 Author: Yang Luo <[email protected]> AuthorDate: Sat Sep 5 12:20:33 2026 +0800 fix(connections): rewrite an agent's entries when Gateway moves or changes port --- connector/connectors/amap.json | 1 - connector/connectors/feishu.json | 1 - controllers/agent.go | 7 +++++ mcpproxy/gateway.go | 24 ++++++++++++---- mcpproxy/mcpproxy.go | 18 ++++++++++++ object/connection.go | 29 +++++++++++++++++--- object/connection_install.go | 59 ++++++++++++++++++++++++++++++++++++---- 7 files changed, 123 insertions(+), 16 deletions(-) diff --git a/connector/connectors/amap.json b/connector/connectors/amap.json index 1d3b1d2..639f7e7 100644 --- a/connector/connectors/amap.json +++ b/connector/connectors/amap.json @@ -5,7 +5,6 @@ "icon": "amap.com", "description": "Search places, plan driving and walking routes, check the weather, and turn an address into coordinates.", "docsUrl": "https://lbs.amap.com/api/mcp-server/summary";, - "unverified": true, "auth": { "kind": "token", "fields": [ diff --git a/connector/connectors/feishu.json b/connector/connectors/feishu.json index cd32b0d..6a14b76 100644 --- a/connector/connectors/feishu.json +++ b/connector/connectors/feishu.json @@ -5,7 +5,6 @@ "icon": "feishu.cn", "description": "Work with Lark from inside an agent: messages and groups, documents, spreadsheets, calendars and contacts, through your own company's app.", "docsUrl": "https://github.com/larksuite/lark-openapi-mcp";, - "unverified": true, "auth": { "kind": "token", "fields": [ diff --git a/controllers/agent.go b/controllers/agent.go index f2e5903..1acbf68 100644 --- a/controllers/agent.go +++ b/controllers/agent.go @@ -82,6 +82,13 @@ func (c *ApiController) GetAgents() { return } + // A connection's entries name this Gateway's address and program, so they go + // stale the same way a hook does. Repairing them here means the fix lands + // wherever the agents are listed, without a page of its own to visit. + if err := object.EnsureConnectionsCurrent(); err != nil { + beego.Error("some connections could not be written again:", err) + } + result := make([]*discoveredAgent, 0, len(installations)) baseUrls := map[string]string{} for _, installation := range installations { diff --git a/mcpproxy/gateway.go b/mcpproxy/gateway.go index 1930d4f..486f167 100644 --- a/mcpproxy/gateway.go +++ b/mcpproxy/gateway.go @@ -19,7 +19,6 @@ import ( "encoding/json" "fmt" "net/http" - "net/url" "time" "github.com/apache/casbin-gateway/agentmonitor" @@ -158,8 +157,23 @@ func toolKey(connection string, tool string) string { return "mcp__" + connection + "__" + tool } -// GatewayUrl is the address a patched agent is given for the Gateway on this -// machine. -func GatewayUrl(port string) string { - return (&url.URL{Scheme: "http", Host: "127.0.0.1:" + port}).String() +// report records one call through this connection. Nothing waits on it: a +// record that cannot be written is worth less than the session it would have +// held up, which is how every other reporter here behaves. +func (g *gatewayClient) report(tool string, outcome string, detail string) { + if g.options.AgentId == "" || g.options.Token == "" { + return + } + + _, _ = g.post("/api/add-agent-record", map[string]any{ + "agent": g.options.AgentId, + "user": g.options.Owner, + "eventType": "tool", + "action": "call", + "outcome": outcome, + "toolName": toolKey(g.options.Connection, tool), + "mcpServer": g.options.Connection, + "mcpTool": tool, + "detail": detail, + }, decideTimeout) } diff --git a/mcpproxy/mcpproxy.go b/mcpproxy/mcpproxy.go index a49a35d..e89bc2a 100644 --- a/mcpproxy/mcpproxy.go +++ b/mcpproxy/mcpproxy.go @@ -101,6 +101,9 @@ func Serve(in *os.File, out *os.File, options Options) error { if err := server.Start(proxy.emit); err != nil { return err } + // Records are written beside the session rather than in it, so the last few + // would be lost to the exit that follows this return. + defer proxy.reporting.Wait() return proxy.pump(in) } @@ -113,6 +116,18 @@ type proxy struct { // writing serializes what reaches the agent: replies the upstream produced // and refusals written here are both messages on one stream. writing sync.Mutex + // reporting counts the records still in flight, so the session waits for + // them instead of the process exiting from under them. + reporting sync.WaitGroup +} + +// reportCall records one forwarded call without the session waiting on it. +func (p *proxy) reportCall(tool string) { + p.reporting.Add(1) + go func() { + defer p.reporting.Done() + p.client.report(tool, "attempted", "") + }() } // message is only what the proxy has to look at. Everything else about a @@ -144,11 +159,14 @@ func (p *proxy) pump(in *os.File) error { if err := json.Unmarshal(forwarded, &parsed); err == nil && parsed.Method == "tools/call" { allowed, reason := p.client.allowTool(parsed.Params.Name) if !allowed { + // The refusal is already recorded by the endpoint that decided + // it, so only the calls that go through are reported here. if err := p.refuse(parsed.Id, reason); err != nil { return err } continue } + p.reportCall(parsed.Params.Name) } if err := p.server.Send(forwarded); err != nil { diff --git a/object/connection.go b/object/connection.go index c616097..2d3034e 100644 --- a/object/connection.go +++ b/object/connection.go @@ -29,10 +29,9 @@ import ( // single thing rather than a copy per agent: connecting writes it into every // agent that was picked, and disconnecting takes it out of all of them. // -// The rendered credential still lands in each agent's own configuration file, -// because that is where its MCP client reads a server from. Keeping the secret -// out of those files needs Gateway to proxy the server itself, which it does -// not do yet. +// What is written into an agent is Gateway standing in front of the server, so +// the credential here is the only copy: it is handed to the proxy over loopback +// when a session starts and never reaches the agent's own file. type Connection struct { Owner string `xorm:"varchar(100) notnull pk" json:"owner"` // Name is the connector's catalog id: one connection per connector. @@ -46,6 +45,13 @@ type Connection struct { Credentials map[string]string `xorm:"mediumtext json" json:"credentials"` // Agents is every agent id this connection's MCP server was written into. Agents []string `xorm:"mediumtext json" json:"agents"` + // Endpoint and Executable are the Gateway the written entries name. A + // Gateway whose port changed, or that moved on disk, leaves every one of + // those entries starting a program that is not there or reporting to a port + // nothing answers on; comparing these is what notices, so the entries can be + // written again rather than failing silently in each agent. + Endpoint string `xorm:"varchar(255)" json:"endpoint"` + Executable string `xorm:"varchar(500)" json:"executable"` } func (connection *Connection) GetId() string { @@ -135,6 +141,21 @@ func GetConnections(owner string) ([]*Connection, error) { return connections, nil } +// getAllConnections is every connection on this machine, whoever owns it. The +// repair pass uses it: an entry naming a Gateway that has moved is stale for +// its owner as much as for anybody else. +func getAllConnections() ([]*Connection, error) { + connections := []*Connection{} + if err := ormer.Engine.Find(&connections); err != nil { + return nil, err + } + + for _, connection := range connections { + decryptConnection(connection) + } + return connections, nil +} + func GetConnection(owner string, name string) (*Connection, error) { connection := Connection{Owner: owner, Name: name} existed, err := ormer.Engine.Get(&connection) diff --git a/object/connection_install.go b/object/connection_install.go index f62d209..8c7c654 100644 --- a/object/connection_install.go +++ b/object/connection_install.go @@ -67,6 +67,14 @@ func InstallConnection(connection *Connection, agentIds []string) ([]*agentconfi planned = append(planned, removeFrom(connection, found.Server.Name, dropped(connection.Agents, agentIds))...) connection.Agents = installed(planned, agentIds) + // Remembering which Gateway the entries name is what lets a later start + // notice that they are stale, see EnsureConnectionsCurrent. + if executable, err := agentpatch.GatewayExecutable(); err == nil { + connection.Executable = executable + } + if gatewayUrl, err := agentpatch.GatewayBaseUrl(); err == nil { + connection.Endpoint = gatewayUrl + } if err := SaveConnection(connection); err != nil { return nil, err } @@ -99,13 +107,16 @@ func proxyRequest(connection *Connection, found connector.Connector, agentId str Name: found.Server.Name, Transport: agentconfig.TransportStdio, Command: executable, + // "--flag=value" rather than two arguments: a machine-wide installation + // has no owner, and an empty value written as its own argument makes the + // flag swallow the next one instead. Args: []string{ mcpproxy.Subcommand, - "--connection", connection.Name, - "--agent", agentId, - "--owner", connection.Owner, - "--gateway-url", gatewayUrl, - "--token", token, + "--connection=" + connection.Name, + "--agent=" + agentId, + "--owner=" + connection.Owner, + "--gateway-url=" + gatewayUrl, + "--token=" + token, }, Overwrite: true, }, nil @@ -204,3 +215,41 @@ func installed(planned []*agentconfig.PlanItem, agentIds []string) []string { sort.Strings(found) return found } + +// EnsureConnectionsCurrent writes again every connection whose entries name a +// Gateway this one is not. An installation's hooks get the same treatment from +// agentpatch, and for the same reason: a port that changed or a program that +// moved leaves the entry naming something that is no longer there, and nothing +// else on the machine would ever notice. +// +// It compares two strings per connection and touches no file when they match, +// so it is cheap enough to run wherever the agents are listed. +func EnsureConnectionsCurrent() error { + executable, err := agentpatch.GatewayExecutable() + if err != nil { + return err + } + gatewayUrl, err := agentpatch.GatewayBaseUrl() + if err != nil { + return err + } + + connections, err := getAllConnections() + if err != nil { + return err + } + + var failed error + for _, connection := range connections { + if len(connection.Agents) == 0 { + continue + } + if connection.Endpoint == gatewayUrl && connection.Executable == executable { + continue + } + if _, err := InstallConnection(connection, connection.Agents); err != nil { + failed = fmt.Errorf("connection %s: %w", connection.GetId(), err) + } + } + return failed +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
