libp2p.host package

Submodules

libp2p.host.basic_host module

class libp2p.host.basic_host.BasicHost(network: INetworkService, enable_mDNS: bool = False, enable_upnp: bool = False, enable_autotls: bool = False, bootstrap: list[str] | None = None, default_protocols: OrderedDict[TProtocol, StreamHandlerFn] | None = None, negotiate_timeout: int = 30, resource_manager: ResourceManager | None = None, psk: str | None = None, *, bootstrap_allow_ipv6: bool = False, bootstrap_dns_timeout: float = 10.0, bootstrap_dns_max_retries: int = 3, announce_addrs: Sequence[multiaddr.Multiaddr] | None = None)

Bases: IHost

BasicHost is a wrapper of a INetwork implementation.

It performs protocol negotiation on a stream with multistream-select right after a stream is initialized.

bootstrap: BootstrapDiscovery | None

async close() → None: Close the host and all underlying connections and services.

async connect(peer_info: PeerInfo) → None

Ensure there is a connection between this host and the peer with given peer_info.peer_id. connect will absorb the addresses in peer_info into its internal peerstore. If there is not an active connection, connect will issue a dial, and block until a connection is opened, or an error is returned.

This method ensures the connection is fully established and ready for streams before returning, including: - QUIC handshake completion - Muxer initialization - Connection registration in swarm - Stream handler readiness

Parameters:: peer_info (peer.peerinfo.PeerInfo) – peer_info of the peer we want to connect to

async disconnect(peer_id: ID) → None

Disconnect from the specified peer.

Parameters

peer_idID: The identifier of the peer to disconnect from.

get_addrs() → list[Multiaddr]

Return the multiaddr addresses this host advertises to peers.

Behavior (mirrors go-libp2p’s AddrsFactory pipeline):

If announce_addrs was provided at construction time, that list replaces everything — it is treated as a static AddrsFactory in go-libp2p terms. Observed (NAT) addresses are still recorded by ObservedAddrManager (for get_nat_type and future AutoNAT consumers) but are not emitted here, since the caller has explicitly chosen which addresses to advertise.
Otherwise the set of raw transport addresses is augmented with externally observed addresses that have been confirmed by enough distinct peer groups (see ACTIVATION_THRESHOLD), then the /p2p/{peer_id} suffix is appended to each.

Use get_transport_addrs() for the raw transport addresses without any observed-address augmentation or /p2p suffix.

get_connected_peers() → list[ID]

Returns:: all the ids of peers this host is currently connected to

get_id() → ID

Returns:: peer_id of host

get_live_peers() → list[ID]

Returns a list of currently connected peer IDs.

Returns:: List of peer IDs that have active connections

get_mux() → Multiselect

Returns:: mux instance of host

get_nat_type() → tuple[NATDeviceType, NATDeviceType]

Return the classified NAT device type for TCP and UDP transports.

Thin pass-through to libp2p.host.observed_addr_manager.ObservedAddrManager.get_nat_type(), which infers NAT behaviour from the distribution of externally observed addresses reported through Identify. Matches go-libp2p’s host.getNATType() algorithm.

Note

Experimental API. Intended primarily for AutoNAT / hole-punch consumers; the return values, thresholds, and method name may evolve as those subsystems land in py-libp2p.

Returns:: (tcp_nat_type, udp_nat_type), each one of NATDeviceType.

get_network() → INetworkService

Returns:: network instance of host

get_peer_connection_info(peer_id: ID) → list[INetConn] | None

Get connection information for a specific peer if connected.

Parameters:: peer_id – ID of the peer to get info for
Returns:: Connection object if peer is connected, None otherwise

get_peerstore() → IPeerStore

Returns:: peerstore of the host (same one as in its network instance)

get_private_key() → PrivateKey

Retrieve the private key of the host.

Returns

PrivateKey: The private key belonging to the host.

get_public_key() → PublicKey

Retrieve the public key of the host.

Returns

PublicKey: The public key belonging to the host.

get_transport_addrs() → list[Multiaddr]: Return the raw multiaddr addresses this host is listening to, without the /p2p/{peer_id} suffix.

async initiate_autotls_procedure(public_ip: str | None = None) → None

Run the AutoTLS certificate provisioning flow for this host.

If a cached ACME certificate already exists on disk, it is loaded and validated and procedure exists early. Otherwise the method performs the full AutoTLS flow:

create or load an ACME account bound to the host’s identity key
initiate a certificate order
obtain a DNS-01 challenge
discover a publicly reachable IPv4 address from the host’s listen addrs
register the challenge with the AutoTLS broker
wait for DNS propagation
finalize the order and fetch the certificate

Only publicly reachable IPv4 addresses are considered valid for AutoTLS. If no such address can be determined, the procedure fails.

Parameters:: public_ip – Optional externally known public IPv4 address. If not provided, the address is inferred from the host’s transport addresses.
Returns:: None
Raises:: RuntimeError – if no publicly reachable IPv4 address can be determined for DNS challenge registration.

is_peer_connected(peer_id: ID) → bool

Check if a specific peer is currently connected.

Parameters:: peer_id – ID of the peer to check
Returns:: True if peer has an active connection, False otherwise

mDNS: MDNSDiscovery | None

multiselect: Multiselect

multiselect_client: MultiselectClient

async new_stream(peer_id: ID, protocol_ids: Sequence[TProtocol]) → INetStream

Parameters:

peer_id – peer_id that host is connecting
protocol_ids – available protocol ids to use for stream

Returns:

stream: new stream created

peerstore: IPeerStore

remove_stream_handler(protocol_id: TProtocol) → None

Remove the stream handler for the given protocol_id.

Parameters:: protocol_id – protocol id to remove

run(listen_addrs: Sequence[Multiaddr], *, task_status: Any = TASK_STATUS_IGNORED) → AbstractAsyncContextManager[None]

Run the host instance and listen to listen_addrs.

Parameters:

listen_addrs – a sequence of multiaddrs that we want to listen to
task_status – task status for trio nursery compatibility (ignored)

async send_command(peer_id: ID, command: str, response_timeout: int = 30) → list[str]

Send a multistream-select command to the specified peer and return the response.

Parameters:

peer_id – peer_id that host is connecting
command – supported multistream-select command (e.g., “ls)

Raises:

StreamFailure – If the stream cannot be opened or negotiation fails

Returns:

list of strings representing the response from peer.

set_stream_handler(protocol_id: TProtocol, stream_handler: Callable[[object], Awaitable[None]]) → None

Set stream handler for given protocol_id

Parameters:

protocol_id – protocol id used on stream
stream_handler – a stream handler function

async upgrade_inbound_connection(raw_conn: IRawConnection, maddr: Multiaddr) → IMuxedConn

Upgrade a raw inbound connection using the underlying network.

Parameters:

raw_conn – The inbound raw connection to upgrade.
maddr – The multiaddress this connection arrived on.

Raises:

SwarmException – raised when security or muxer upgrade fails

Returns:

network connection with security and multiplexing established

async upgrade_outbound_connection(raw_conn: IRawConnection, peer_id: ID) → INetConn

Upgrade a raw outbound connection for the peer_id using the underlying network.

Parameters:

raw_conn – The raw connection to upgrade.
peer_id – The peer this connection is to.

Raises:

SwarmException – raised when security or muxer upgrade fails

Returns:

network connection with security and multiplexing established

upnp: UpnpManager | None

libp2p.host.defaults module

libp2p.host.defaults.get_default_protocols(host: IHost) → OrderedDict[TProtocol, StreamHandlerFn]

libp2p.host.exceptions module

exception libp2p.host.exceptions.ConnectionFailure: Bases: HostException

exception libp2p.host.exceptions.HostException

Bases: BaseLibp2pError

A generic exception in IHost.

exception libp2p.host.exceptions.StreamFailure: Bases: HostException

libp2p.host.observed_addr_manager module

Parameters

conn:: The network connection the observation came from.
observed_addr:: The address the remote peer sees us as.
local_addrs:: Our current listen/transport addresses (without /p2p/…).

remove_conn(conn: INetConn) → None: Clean up observations when a connection is closed.

libp2p.host.observed_addr_manager.extract_thin_waist(maddr: Multiaddr) → tuple[Multiaddr, str] | None

Split maddr into a thin-waist prefix and the remaining suffix.

The thin waist is the IP + transport portion, e.g. /ip4/1.2.3.4/tcp/4001. Everything after (/ws, /p2p/Qm…, …) is returned as the rest string.

Returns None when the address does not contain a recognisable thin-waist prefix.

libp2p.host.observed_addr_manager.has_consistent_transport(tw_a: Multiaddr, tw_b: Multiaddr) → bool: Check both thin waists use the same protocol codes (IP + transport).

libp2p.host.observed_addr_manager.is_valid_observation(observed: Multiaddr) → bool: Reject loopback, relay, NAT64, and non-thin-waist observations.

libp2p.host.observed_addr_manager.observer_group(remote_addr: tuple[str, int]) → str

Compute a grouping key from the observer’s remote address.

IPv4: full IP string. IPv6: /56 prefix. Multiple connections from the same group count as a single observer.

libp2p.host.ping module

class libp2p.host.ping.PingService(host: IHost)

Bases: object

PingService executes pings and returns RTT in miliseconds.

async ping(peer_id: ID, ping_amt: int = 1) → list[int]

async libp2p.host.ping.handle_ping(stream: INetStream) → None: Respond to incoming ping requests until one side errors or closes the stream.

libp2p.host.routed_host module

class libp2p.host.routed_host.RoutedHost(network: INetworkService, router: IPeerRouting, enable_mDNS: bool = False, enable_upnp: bool = False, enable_autotls: bool = False, bootstrap: list[str] | None = None, resource_manager: ResourceManager | None = None, *, bootstrap_allow_ipv6: bool = False, bootstrap_dns_timeout: float = 10.0, bootstrap_dns_max_retries: int = 3, announce_addrs: Sequence[multiaddr.Multiaddr] | None = None)

Bases: BasicHost

RoutedHost is a p2p Host that includes a routing system.

This allows the Host to find the addresses for peers when it does not have them.

async connect(peer_info: PeerInfo) → None

Ensure there is a connection between this host and the peer with given peer_info.peer_id. See (basic_host).connect for more information.

RoutedHost’s Connect differs in that if the host has no addresses for a given peer, it will use its routing system to try to find some.

Parameters:: peer_info (peer.peerinfo.PeerInfo) – peer_info of the peer we want to connect to

libp2p.host package

Subpackages

Submodules

libp2p.host.basic_host module

Parameters

Returns

Returns

libp2p.host.defaults module

libp2p.host.exceptions module

libp2p.host.observed_addr_manager module

Parameters

libp2p.host.ping module

libp2p.host.routed_host module

Module contents